Merge branch 'main' into contractServiceQueriesSimpleFeesTestRefactor

Author: Evdokia-Georgieva

.snyk

@@ -18,4 +18,6 @@ exclude:
     - >-
       platform-sdk/consensus-sloth/src/testFixtures/java/org/hiero/sloth/fixtures/container/docker/DockerManager.java
     - >-
-      platform-sdk/consensus-sloth/src/testFixtures/java/org/hiero/sloth/fixtures/container/docker/platform/NodeCommunicationService.java
+      platform-sdk/consensus-sloth/src/testFixtures/java/org/hiero/sloth/fixtures/container/docker/platform/NodeCommunicationService.java
+    - >-
+      platform-sdk/consensus-gossip-impl/src/main/java/org/hiero/consensus/gossip/impl/network/PeerConnectionServer.java

platform-sdk/consensus-gossip-impl/src/main/java/org/hiero/consensus/gossip/impl/gossip/sync/SyncInputStream.java

@@ -1,8 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.consensus.gossip.impl.gossip.sync;
 
-import static org.hiero.consensus.io.extendable.ExtendableInputStream.extendInputStream;
-
 import com.swirlds.config.api.Configuration;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.BufferedInputStream;
@@ -11,25 +9,38 @@
 import java.util.zip.InflaterInputStream;
 import org.hiero.base.io.streams.SerializableDataInputStream;
 import org.hiero.consensus.gossip.config.SocketConfig;
-import org.hiero.consensus.io.extendable.extensions.CountingStreamExtension;
-
+import org.hiero.consensus.io.counting.ByteCounter;
+import org.hiero.consensus.io.counting.CounterType;
+import org.hiero.consensus.io.counting.CountingInputStream;
+
+/**
+ * A {@link SerializableDataInputStream} that counts the number of bytes read from it and optionally decompresses
+ * the data using gzip compression.
+ */
 public class SyncInputStream extends SerializableDataInputStream {
 
-    private final CountingStreamExtension syncByteCounter;
+    private final ByteCounter byteCounter;
 
-    private SyncInputStream(@NonNull final InputStream in, @NonNull final CountingStreamExtension syncByteCounter) {
+    private SyncInputStream(@NonNull final InputStream in, @NonNull final ByteCounter byteCounter) {
         super(in);
-        this.syncByteCounter = syncByteCounter;
+        this.byteCounter = byteCounter;
     }
 
+    /**
+     * Create a new {@link SyncInputStream} that optionally decompresses the data using gzip compression and
+     * counts the number of bytes read from it.
+     *
+     * @param configuration the configuration to use to determine whether to use gzip compression
+     * @param in the input stream to read from
+     * @param bufferSize the buffer size to use when reading from the input stream
+     * @return a new {@link SyncInputStream}
+     */
     public static SyncInputStream createSyncInputStream(
             @NonNull final Configuration configuration, @NonNull final InputStream in, final int bufferSize) {
 
-        final CountingStreamExtension syncCounter = new CountingStreamExtension();
-
         final boolean compress = configuration.getConfigData(SocketConfig.class).gzipCompression();
 
-        final InputStream meteredStream = extendInputStream(in, syncCounter);
+        final CountingInputStream meteredStream = new CountingInputStream(in, CounterType.THREAD_SAFE);
 
         final InputStream wrappedStream;
         if (compress) {
@@ -38,10 +49,15 @@ public static SyncInputStream createSyncInputStream(
             wrappedStream = new BufferedInputStream(meteredStream, bufferSize);
         }
 
-        return new SyncInputStream(wrappedStream, syncCounter);
+        return new SyncInputStream(wrappedStream, meteredStream.byteCounter());
     }
 
-    public CountingStreamExtension getSyncByteCounter() {
-        return syncByteCounter;
+    /**
+     * Get the byte counter that counts the number of bytes read from this stream.
+     *
+     * @return the {@link ByteCounter}
+     */
+    public ByteCounter byteCounter() {
+        return byteCounter;
     }
 }

platform-sdk/consensus-gossip-impl/src/main/java/org/hiero/consensus/gossip/impl/gossip/sync/SyncMetrics.java

@@ -9,7 +9,6 @@
 import static com.swirlds.metrics.api.Metrics.PLATFORM_CATEGORY;
 
 import com.swirlds.base.time.Time;
-import com.swirlds.base.units.UnitConstants;
 import com.swirlds.metrics.api.IntegerGauge;
 import com.swirlds.metrics.api.Metrics;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -25,17 +24,13 @@
 import org.hiero.consensus.gossip.impl.gossip.shadowgraph.ShadowgraphSynchronizer;
 import org.hiero.consensus.gossip.impl.gossip.shadowgraph.SyncPhase;
 import org.hiero.consensus.gossip.impl.gossip.shadowgraph.SyncResult;
-import org.hiero.consensus.gossip.impl.gossip.shadowgraph.SyncTiming;
-import org.hiero.consensus.gossip.impl.network.Connection;
 import org.hiero.consensus.gossip.impl.network.PeerInfo;
 import org.hiero.consensus.metrics.RunningAverageMetric;
 import org.hiero.consensus.metrics.extensions.CountPerSecond;
 import org.hiero.consensus.metrics.extensions.PhaseTimer;
 import org.hiero.consensus.metrics.extensions.PhaseTimerBuilder;
 import org.hiero.consensus.metrics.statistics.AverageAndMax;
-import org.hiero.consensus.metrics.statistics.AverageAndMaxTimeStat;
 import org.hiero.consensus.metrics.statistics.AverageStat;
-import org.hiero.consensus.metrics.statistics.AverageTimeStat;
 import org.hiero.consensus.metrics.statistics.MaxStat;
 import org.hiero.consensus.model.hashgraph.EventWindow;
 import org.hiero.consensus.model.node.NodeId;
@@ -45,11 +40,6 @@
  */
 public class SyncMetrics {
 
-    private static final RunningAverageMetric.Config AVG_BYTES_PER_SEC_SYNC_CONFIG = new RunningAverageMetric.Config(
-                    PLATFORM_CATEGORY, "bytes_per_sec_sync")
-            .withDescription("average number of bytes per second transferred during a sync");
-    private final RunningAverageMetric avgBytesPerSecSync;
-
     private static final RunningAverageMetric.Config TIPS_PER_SYNC_CONFIG = new RunningAverageMetric.Config(
                     INTERNAL_CATEGORY, "tips_per_sync")
             .withDescription("the average number of tips per sync at the start of each sync")
@@ -196,12 +186,6 @@ public class SyncMetrics {
 
     private final AverageStat syncIndicatorDiff;
     private final AverageStat eventRecRate;
-    private final AverageTimeStat avgSyncDuration1;
-    private final AverageTimeStat avgSyncDuration2;
-    private final AverageTimeStat avgSyncDuration3;
-    private final AverageTimeStat avgSyncDuration4;
-    private final AverageTimeStat avgSyncDuration5;
-    private final AverageAndMaxTimeStat avgSyncDuration;
     private final AverageStat knownSetSize;
     private final AverageAndMax avgEventsPerSyncSent;
     private final AverageAndMax avgEventsPerSyncRec;
@@ -227,7 +211,6 @@ public class SyncMetrics {
     public SyncMetrics(final Metrics metrics, final Time time, final List<PeerInfo> peers) {
         this.metrics = Objects.requireNonNull(metrics);
         this.time = Objects.requireNonNull(time);
-        avgBytesPerSecSync = metrics.getOrCreate(AVG_BYTES_PER_SEC_SYNC_CONFIG);
         callSyncsPerSecond = new CountPerSecond(metrics, CALL_SYNCS_PER_SECOND_CONFIG);
         recSyncsPerSecond = new CountPerSecond(metrics, REC_SYNCS_PER_SECOND_CONFIG);
         tipsPerSync = metrics.getOrCreate(TIPS_PER_SYNC_CONFIG);
@@ -258,13 +241,6 @@ public SyncMetrics(final Metrics metrics, final Time time, final List<PeerInfo>
         broadcastDisabledDueToLag = metrics.getOrCreate(BROADCAST_DISABLED_DUE_TO_LAG_CONFIG);
         broadcastDisabledDueToOverload = metrics.getOrCreate(BROADCAST_DISABLED_DUE_TO_OVERLOAD_CONFIG);
 
-        avgSyncDuration = new AverageAndMaxTimeStat(
-                metrics,
-                ChronoUnit.SECONDS,
-                INTERNAL_CATEGORY,
-                "sec_per_sync",
-                "duration of average successful sync (in seconds)");
-
         avgEventsPerSyncSent = new AverageAndMax(
                 metrics, PLATFORM_CATEGORY, "ev_per_syncS", "number of events sent per successful sync", FORMAT_8_1);
         avgEventsPerSyncRec = new AverageAndMax(
@@ -289,37 +265,6 @@ public SyncMetrics(final Metrics metrics, final Time time, final List<PeerInfo>
                 FORMAT_8_1,
                 AverageStat.WEIGHT_VOLATILE);
 
-        avgSyncDuration1 = new AverageTimeStat(
-                metrics,
-                ChronoUnit.SECONDS,
-                INTERNAL_CATEGORY,
-                "sec_per_sync1",
-                "duration of step 1 of average successful sync (in seconds)");
-        avgSyncDuration2 = new AverageTimeStat(
-                metrics,
-                ChronoUnit.SECONDS,
-                INTERNAL_CATEGORY,
-                "sec_per_sync2",
-                "duration of step 2 of average successful sync (in seconds)");
-        avgSyncDuration3 = new AverageTimeStat(
-                metrics,
-                ChronoUnit.SECONDS,
-                INTERNAL_CATEGORY,
-                "sec_per_sync3",
-                "duration of step 3 of average successful sync (in seconds)");
-        avgSyncDuration4 = new AverageTimeStat(
-                metrics,
-                ChronoUnit.SECONDS,
-                INTERNAL_CATEGORY,
-                "sec_per_sync4",
-                "duration of step 4 of average successful sync (in seconds)");
-        avgSyncDuration5 = new AverageTimeStat(
-                metrics,
-                ChronoUnit.SECONDS,
-                INTERNAL_CATEGORY,
-                "sec_per_sync5",
-                "duration of step 5 of average successful sync (in seconds)");
-
         knownSetSize = new AverageStat(
                 metrics,
                 PLATFORM_CATEGORY,
@@ -382,30 +327,6 @@ public void eventsReceived(final long nanosStart, final int numberReceived) {
         eventRecRate.update(Math.round(numberReceived / seconds));
     }
 
-    /**
-     * Record all stats related to sync timing
-     *
-     * @param timing object that holds the timing data
-     * @param conn   the sync connections
-     */
-    public void recordSyncTiming(final SyncTiming timing, final Connection conn) {
-        avgSyncDuration1.update(timing.getTimePoint(0), timing.getTimePoint(1));
-        avgSyncDuration2.update(timing.getTimePoint(1), timing.getTimePoint(2));
-        avgSyncDuration3.update(timing.getTimePoint(2), timing.getTimePoint(3));
-        avgSyncDuration4.update(timing.getTimePoint(3), timing.getTimePoint(4));
-        avgSyncDuration5.update(timing.getTimePoint(4), timing.getTimePoint(5));
-
-        avgSyncDuration.update(timing.getTimePoint(0), timing.getTimePoint(5));
-        final double syncDurationSec = timing.getPointDiff(5, 0) * UnitConstants.NANOSECONDS_TO_SECONDS;
-        final double speed = Math.max(
-                        conn.getDis().getSyncByteCounter().getCount(),
-                        conn.getDos().getSyncByteCounter().getCount())
-                / syncDurationSec;
-
-        // set the bytes/sec speed of the sync currently measured
-        avgBytesPerSecSync.update(speed);
-    }
-
     /**
      * Records the size of the known set during a sync. This is the most compute intensive part of the sync, so this is
      * useful information to validate sync performance.

platform-sdk/consensus-gossip-impl/src/main/java/org/hiero/consensus/gossip/impl/gossip/sync/SyncOutputStream.java

@@ -1,8 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.consensus.gossip.impl.gossip.sync;
 
-import static org.hiero.consensus.io.extendable.ExtendableOutputStream.extendOutputStream;
-
 import com.swirlds.config.api.Configuration;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.BufferedOutputStream;
@@ -11,29 +9,38 @@
 import java.util.zip.DeflaterOutputStream;
 import org.hiero.base.io.streams.SerializableDataOutputStream;
 import org.hiero.consensus.gossip.config.SocketConfig;
-import org.hiero.consensus.io.extendable.extensions.CountingStreamExtension;
+import org.hiero.consensus.io.counting.ByteCounter;
+import org.hiero.consensus.io.counting.CounterType;
+import org.hiero.consensus.io.counting.CountingOutputStream;
 
+/**
+ * A {@link SerializableDataOutputStream} that counts the number of bytes written to it and optionally compresses
+ * the data using gzip compression.
+ */
 public class SyncOutputStream extends SerializableDataOutputStream {
-    private final CountingStreamExtension syncByteCounter;
-    private final CountingStreamExtension connectionByteCounter;
 
-    protected SyncOutputStream(
-            @NonNull final OutputStream out,
-            @NonNull final CountingStreamExtension syncByteCounter,
-            @NonNull final CountingStreamExtension connectionByteCounter) {
+    private final ByteCounter connectionByteCounter;
+
+    protected SyncOutputStream(@NonNull final OutputStream out, @NonNull final ByteCounter connectionByteCounter) {
         super(out);
-        this.syncByteCounter = syncByteCounter;
         this.connectionByteCounter = connectionByteCounter;
     }
 
+    /**
+     * Create a new {@link SyncOutputStream} that optionally compresses the data using gzip compression and
+     * counts the number of bytes written to it.
+     *
+     * @param configuration the configuration to use to determine whether to use gzip compression
+     * @param out the output stream to write to
+     * @param bufferSize the buffer size to use when writing to the output stream
+     * @return a new {@link SyncOutputStream}
+     */
     public static SyncOutputStream createSyncOutputStream(
             @NonNull final Configuration configuration, @NonNull final OutputStream out, final int bufferSize) {
-        final CountingStreamExtension syncByteCounter = new CountingStreamExtension();
-        final CountingStreamExtension connectionByteCounter = new CountingStreamExtension();
 
         final boolean compress = configuration.getConfigData(SocketConfig.class).gzipCompression();
 
-        final OutputStream meteredStream = extendOutputStream(out, connectionByteCounter);
+        final CountingOutputStream meteredStream = new CountingOutputStream(out, CounterType.THREAD_SAFE);
 
         final OutputStream wrappedStream;
         if (compress) {
@@ -44,14 +51,15 @@ public static SyncOutputStream createSyncOutputStream(
         }
 
         // we write the data to the buffer first, for efficiency
-        return new SyncOutputStream(wrappedStream, syncByteCounter, connectionByteCounter);
-    }
-
-    public CountingStreamExtension getSyncByteCounter() {
-        return syncByteCounter;
+        return new SyncOutputStream(wrappedStream, meteredStream.byteCounter());
     }
 
-    public CountingStreamExtension getConnectionByteCounter() {
+    /**
+     * Get the connection byte counter that counts the number of bytes written to this stream.
+     *
+     * @return the {@link ByteCounter}
+     */
+    public ByteCounter connectionByteCounter() {
         return connectionByteCounter;
     }
 }

platform-sdk/consensus-gossip-impl/src/main/java/org/hiero/consensus/gossip/impl/network/NetworkMetrics.java

@@ -194,7 +194,7 @@ public void update() {
         for (final Iterator<Connection> iterator = connections.iterator(); iterator.hasNext(); ) {
             final Connection conn = iterator.next();
             if (conn != null) {
-                final long bytesSent = conn.getDos().getConnectionByteCounter().getAndResetCount();
+                final long bytesSent = conn.getDos().connectionByteCounter().getAndReset();
                 totalBytesSent += bytesSent;
                 final NodeId otherId = conn.getOtherId();
 

platform-sdk/consensus-gossip-impl/src/main/java/org/hiero/consensus/gossip/impl/network/SocketConnection.java

@@ -182,8 +182,7 @@ public void initForSync() throws IOException {
         }
 
         /* track the number of bytes written and read during a sync */
-        getDis().getSyncByteCounter().resetCount();
-        getDos().getSyncByteCounter().resetCount();
+        getDis().byteCounter().getAndReset();
         final SocketConfig socketConfig = configuration.getConfigData(SocketConfig.class);
         this.setTimeout(socketConfig.timeoutSyncClientSocket());
     }

platform-sdk/consensus-gossip-impl/src/test/java/org/hiero/consensus/gossip/impl/network/SocketConnectionTests.java

@@ -23,7 +23,7 @@
 import org.hiero.base.io.exceptions.BadIOException;
 import org.hiero.consensus.gossip.impl.gossip.sync.SyncInputStream;
 import org.hiero.consensus.gossip.impl.gossip.sync.SyncOutputStream;
-import org.hiero.consensus.io.extendable.extensions.CountingStreamExtension;
+import org.hiero.consensus.io.counting.ByteCounter;
 import org.hiero.consensus.model.node.NodeId;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -174,7 +174,8 @@ void testConnected_SocketThrows(final boolean outbound) {
 
         initConnection(outbound);
 
-        boolean isConnected = assertDoesNotThrow(conn::connected, "connected() should not thrown any exceptions.");
+        final boolean isConnected =
+                assertDoesNotThrow(conn::connected, "connected() should not thrown any exceptions.");
         assertFalse(isConnected, "connected() should return false when an exception occurs.");
     }
 
@@ -219,10 +220,8 @@ selfId, otherId, mock(ConnectionTracker.class), outbound, socket, dis, null, con
     @ParameterizedTest
     @MethodSource("booleans")
     void testInitForSync(final boolean outbound) throws SocketException {
-        CountingStreamExtension disCounter = mock(CountingStreamExtension.class);
-        CountingStreamExtension dosCounter = mock(CountingStreamExtension.class);
-        when(dis.getSyncByteCounter()).thenReturn(disCounter);
-        when(dos.getSyncByteCounter()).thenReturn(dosCounter);
+        final ByteCounter disCounter = mock(ByteCounter.class);
+        when(dis.byteCounter()).thenReturn(disCounter);
 
         // mock Connection.connected() = true
         when(socket.isClosed()).thenReturn(false);
@@ -232,10 +231,8 @@ void testInitForSync(final boolean outbound) throws SocketException {
         initConnection(outbound);
         assertDoesNotThrow(() -> conn.initForSync(), "An exception should not be thrown when the connection is valid.");
 
-        verify(dis).getSyncByteCounter();
-        verify(dos).getSyncByteCounter();
-        verify(disCounter).resetCount();
-        verify(dosCounter).resetCount();
+        verify(dis).byteCounter();
+        verify(disCounter).getAndReset();
         verify(socket).setSoTimeout(anyInt());
     }