diff --git a/metadata/src/main/java/io/streamstack/metadata/raft/MetadataClient.java b/metadata/src/main/java/io/streamstack/metadata/raft/MetadataClient.java
index c65fa94..eaf664b 100644
--- a/metadata/src/main/java/io/streamstack/metadata/raft/MetadataClient.java
+++ b/metadata/src/main/java/io/streamstack/metadata/raft/MetadataClient.java
@@ -7,6 +7,7 @@
 import com.alipay.sofa.jraft.closure.ReadIndexClosure;
 import com.alipay.sofa.jraft.conf.Configuration;
 import com.alipay.sofa.jraft.entity.PeerId;
+import com.alipay.sofa.jraft.error.RaftError;
 import com.alipay.sofa.jraft.option.CliOptions;
 import com.alipay.sofa.jraft.rpc.impl.cli.CliClientServiceImpl;
 
@@ -181,37 +182,62 @@ public CompletableFuture<Void> readIndex(Runnable read) {
 
     public <T> CompletableFuture<T> readIndex(Supplier<T> read) {
         CompletableFuture<T> future = new CompletableFuture<>();
+        attemptReadIndex(read, 0, future);
+        return future;
+    }
 
-        node.raftNode().readIndex(null, new ReadIndexClosure() {
-            @Override
-            public void run(Status status, long index, byte[] reqCtx) {
-                if (!status.isOk()) {
-                    future.completeExceptionally(new IllegalStateException(status.getErrorMsg()));
-                    return;
-                }
+    private <T> void attemptReadIndex(Supplier<T> read, int attempt, CompletableFuture<T> future) {
+        if (future.isDone()) {
+            return;
+        }
 
-                node.stateMachine().awaitApplied(index).whenCompleteAsync((ignored, error) -> {
-                    if (Objects.nonNull(error)) {
-                        future.completeExceptionally(unwrap(error));
+        try {
+            node.raftNode().readIndex(null, new ReadIndexClosure() {
+                @Override
+                public void run(Status status, long index, byte[] reqCtx) {
+                    if (!status.isOk()) {
+                        if (status.getRaftError() == RaftError.EAGAIN && attempt < maxRetries) {
+                            scheduleReadIndexRetry(read, attempt, future);
+                        } else {
+                            future.completeExceptionally(new IllegalStateException(status.getErrorMsg()));
+                        }
                         return;
                     }
 
-                    try {
-                        future.complete(node.stateMachine().read(read));
-                    } catch (Throwable t) {
-                        Throwable cause = unwrap(t);
+                    node.stateMachine().awaitApplied(index).whenCompleteAsync((ignored, error) -> {
+                        if (Objects.nonNull(error)) {
+                            future.completeExceptionally(unwrap(error));
+                            return;
+                        }
 
-                        if (cause instanceof MetadataException metadataException) {
-                            future.completeExceptionally(MetadataException.toStreamClientException(metadataException));
-                        } else {
-                            future.completeExceptionally(cause);
+                        try {
+                            future.complete(node.stateMachine().read(read));
+                        } catch (Throwable t) {
+                            Throwable cause = unwrap(t);
+
+                            if (cause instanceof MetadataException metadataException) {
+                                future.completeExceptionally(MetadataException.toStreamClientException(metadataException));
+                            } else {
+                                future.completeExceptionally(cause);
+                            }
                         }
-                    }
-                }, scheduler);
-            }
-        });
+                    }, scheduler);
+                }
+            });
+        } catch (Throwable t) {
+            future.completeExceptionally(unwrap(t));
+        }
+    }
 
-        return future;
+    private <T> void scheduleReadIndexRetry(Supplier<T> read, int attempt, CompletableFuture<T> future) {
+        try {
+            scheduler.schedule(
+                () -> attemptReadIndex(read, attempt + 1, future),
+                retrySleepMs,
+                TimeUnit.MILLISECONDS);
+        } catch (RuntimeException e) {
+            future.completeExceptionally(e);
+        }
     }
 
     @Override
diff --git a/stream/src/main/java/io/streamstack/s3/S3Storage.java b/stream/src/main/java/io/streamstack/s3/S3Storage.java
index 3fd2cb8..b6e3fb7 100644
--- a/stream/src/main/java/io/streamstack/s3/S3Storage.java
+++ b/stream/src/main/java/io/streamstack/s3/S3Storage.java
@@ -334,7 +334,11 @@ public CompletableFuture<Void> append(AppendContext context, StreamRecordBatch s
         append0(context, writeRequest, false);
         return cf.whenComplete((nil, ex) -> {
             streamRecord.release();
-            APPEND_STORAGE_LATENCY.record(TimerUtil.timeElapsedSince(startTime, TimeUnit.NANOSECONDS));
+            try {
+                APPEND_STORAGE_LATENCY.record(TimerUtil.timeElapsedSince(startTime, TimeUnit.NANOSECONDS));
+            } catch (RuntimeException metricsError) {
+                LOGGER.warn("Failed to record append storage latency", metricsError);
+            }
         });
     }
 
diff --git a/stream/src/main/java/io/streamstack/s3/S3Stream.java b/stream/src/main/java/io/streamstack/s3/S3Stream.java
index 7646e81..9e83b5d 100644
--- a/stream/src/main/java/io/streamstack/s3/S3Stream.java
+++ b/stream/src/main/java/io/streamstack/s3/S3Stream.java
@@ -243,7 +243,11 @@ public CompletableFuture<AppendResult> append(AppendContext context, RecordBatch
             pendingAppends.add(cf);
             PendingRequestTracker.Handle pendingAppend = PENDING_APPEND_TRACKER.begin();
             return cf.whenComplete((nil, ex) -> {
-                APPEND_STREAM_LATENCY.record(TimerUtil.timeElapsedSince(startTimeNanos, TimeUnit.NANOSECONDS));
+                try {
+                    APPEND_STREAM_LATENCY.record(TimerUtil.timeElapsedSince(startTimeNanos, TimeUnit.NANOSECONDS));
+                } catch (RuntimeException metricsError) {
+                    logger.warn("Failed to record append stream latency", metricsError);
+                }
                 pendingAppends.remove(cf);
                 pendingAppend.close();
             });
diff --git a/stream/src/main/java/io/streamstack/s3/metrics/wrapper/DeltaHistogram.java b/stream/src/main/java/io/streamstack/s3/metrics/wrapper/DeltaHistogram.java
index c9ea119..ff88a4a 100644
--- a/stream/src/main/java/io/streamstack/s3/metrics/wrapper/DeltaHistogram.java
+++ b/stream/src/main/java/io/streamstack/s3/metrics/wrapper/DeltaHistogram.java
@@ -103,6 +103,10 @@ private void update(long candidate, AtomicLong target, BiPredicate<Long, Long> p
     }
 
     public void record(long value) {
+        // Silently drop invalid samples because HdrHistogram only supports non-negative values.
+        if (value < 0) {
+            return;
+        }
         cumulativeCount.increment();
         cumulativeSum.add(value);
         this.recorder.recordValue(value);
diff --git a/stream/src/test/java/io/streamstack/s3/metrics/wrapper/MetricsWrapperTest.java b/stream/src/test/java/io/streamstack/s3/metrics/wrapper/MetricsWrapperTest.java
index a4896ee..d29c59d 100644
--- a/stream/src/test/java/io/streamstack/s3/metrics/wrapper/MetricsWrapperTest.java
+++ b/stream/src/test/java/io/streamstack/s3/metrics/wrapper/MetricsWrapperTest.java
@@ -112,6 +112,19 @@ public void testDeltaHistogram() throws InterruptedException {
 //        Assertions.assertEquals(15000, p50, 1000);
     }
 
+    @Test
+    public void testDeltaHistogramIgnoresNegativeValues() {
+        DeltaHistogram histogram = new DeltaHistogram();
+
+        Assertions.assertDoesNotThrow(() -> histogram.record(-1));
+        Assertions.assertEquals(0, histogram.cumulativeCount());
+        Assertions.assertEquals(0, histogram.cumulativeSum());
+
+        histogram.record(1);
+        Assertions.assertEquals(1, histogram.cumulativeCount());
+        Assertions.assertEquals(1, histogram.cumulativeSum());
+    }
+
     private void mockLinearDataDist(DeltaHistogram histogram, int init, int steps) {
         for (int i = init; i < init + steps; i++) {
             histogram.record(i);