S3: Directly read write volume servers (#7481)

* Lazy Versioning Check, Conditional SSE Entry Fetch, HEAD Request Optimization

* revert

Reverted the conditional versioning check to always check versioning status
Reverted the conditional SSE entry fetch to always fetch entry metadata
Reverted the conditional versioning check to always check versioning status
Reverted the conditional SSE entry fetch to always fetch entry metadata

* Lazy Entry Fetch for SSE, Skip Conditional Header Check

* SSE-KMS headers are present, this is not an SSE-C request (mutually exclusive)

* SSE-C is mutually exclusive with SSE-S3 and SSE-KMS

* refactor

* Removed Premature Mutual Exclusivity Check

* check for the presence of the X-Amz-Server-Side-Encryption header

* not used

* fmt

* directly read write volume servers

* HTTP Range Request Support

* set header

* md5

* copy object

* fix sse

* fmt

* implement sse

* sse continue

* fixed the suffix range bug (bytes=-N for "last N bytes")

* debug logs

* Missing PartsCount Header

* profiling

* url encoding

* test_multipart_get_part

* headers

* debug

* adjust log level

* handle part number

* Update s3api_object_handlers.go

* nil safety

* set ModifiedTsNs

* remove

* nil check

* fix sse header

* same logic as filer

* decode values

* decode ivBase64

* s3: Fix SSE decryption JWT authentication and streaming errors

Critical fix for SSE (Server-Side Encryption) test failures:

1. **JWT Authentication Bug** (Root Cause):
   - Changed from GenJwtForFilerServer to GenJwtForVolumeServer
   - S3 API now uses correct JWT when directly reading from volume servers
   - Matches filer's authentication pattern for direct volume access
   - Fixes 'unexpected EOF' and 500 errors in SSE tests

2. **Streaming Error Handling**:
   - Added error propagation in getEncryptedStreamFromVolumes goroutine
   - Use CloseWithError() to properly communicate stream failures
   - Added debug logging for streaming errors

3. **Response Header Timing**:
   - Removed premature WriteHeader(http.StatusOK) call
   - Let Go's http package write status automatically on first write
   - Prevents header lock when errors occur during streaming

4. **Enhanced SSE Decryption Debugging**:
   - Added IV/Key validation and logging for SSE-C, SSE-KMS, SSE-S3
   - Better error messages for missing or invalid encryption metadata
   - Added glog.V(2) debugging for decryption setup

This fixes SSE integration test failures where encrypted objects
could not be retrieved due to volume server authentication failures.
The JWT bug was causing volume servers to reject requests, resulting
in truncated/empty streams (EOF) or internal errors.

* s3: Fix SSE multipart upload metadata preservation

Critical fix for SSE multipart upload test failures (SSE-C and SSE-KMS):

**Root Cause - Incomplete SSE Metadata Copying**:
The old code only tried to copy 'SeaweedFSSSEKMSKey' from the first
part to the completed object. This had TWO bugs:

1. **Wrong Constant Name** (Key Mismatch Bug):
   - Storage uses: SeaweedFSSSEKMSKeyHeader = 'X-SeaweedFS-SSE-KMS-Key'
   - Old code read: SeaweedFSSSEKMSKey = 'x-seaweedfs-sse-kms-key'
   - Result: SSE-KMS metadata was NEVER copied → 500 errors

2. **Missing SSE-C and SSE-S3 Headers**:
   - SSE-C requires: IV, Algorithm, KeyMD5
   - SSE-S3 requires: encrypted key data + standard headers
   - Old code: copied nothing for SSE-C/SSE-S3 → decryption failures

**Fix - Complete SSE Header Preservation**:
Now copies ALL SSE headers from first part to completed object:

- SSE-C: SeaweedFSSSEIV, CustomerAlgorithm, CustomerKeyMD5
- SSE-KMS: SeaweedFSSSEKMSKeyHeader, AwsKmsKeyId, ServerSideEncryption
- SSE-S3: SeaweedFSSSES3Key, ServerSideEncryption

Applied consistently to all 3 code paths:
1. Versioned buckets (creates version file)
2. Suspended versioning (creates main object with null versionId)
3. Non-versioned buckets (creates main object)

**Why This Is Correct**:
The headers copied EXACTLY match what putToFiler stores during part
upload (lines 496-521 in s3api_object_handlers_put.go). This ensures
detectPrimarySSEType() can correctly identify encrypted multipart
objects and trigger inline decryption with proper metadata.

Fixes: TestSSEMultipartUploadIntegration (SSE-C and SSE-KMS subtests)

* s3: Add debug logging for versioning state diagnosis

Temporary debug logging to diagnose test_versioning_obj_plain_null_version_overwrite_suspended failure.

Added glog.V(0) logging to show:
1. setBucketVersioningStatus: when versioning status is changed
2. PutObjectHandler: what versioning state is detected (Enabled/Suspended/none)
3. PutObjectHandler: which code path is taken (putVersionedObject vs putSuspendedVersioningObject)

This will help identify if:
- The versioning status is being set correctly in bucket config
- The cache is returning stale/incorrect versioning state
- The switch statement is correctly routing to suspended vs enabled handlers

* s3: Enhanced versioning state tracing for suspended versioning diagnosis

Added comprehensive logging across the entire versioning state flow:

PutBucketVersioningHandler:
- Log requested status (Enabled/Suspended)
- Log when calling setBucketVersioningStatus
- Log success/failure of status change

setBucketVersioningStatus:
- Log bucket and status being set
- Log when config is updated
- Log completion with error code

updateBucketConfig:
- Log versioning state being written to cache
- Immediate cache verification after Set
- Log if cache verification fails

getVersioningState:
- Log bucket name and state being returned
- Log if object lock forces VersioningEnabled
- Log errors

This will reveal:
1. If PutBucketVersioning(Suspended) is reaching the handler
2. If the cache update succeeds
3. What state getVersioningState returns during PUT
4. Any cache consistency issues

Expected to show why bucket still reports 'Enabled' after 'Suspended' call.

* s3: Add SSE chunk detection debugging for multipart uploads

Added comprehensive logging to diagnose why TestSSEMultipartUploadIntegration fails:

detectPrimarySSEType now logs:
1. Total chunk count and extended header count
2. All extended headers with 'sse'/'SSE'/'encryption' in the name
3. For each chunk: index, SseType, and whether it has metadata
4. Final SSE type counts (SSE-C, SSE-KMS, SSE-S3)

This will reveal if:
- Chunks are missing SSE metadata after multipart completion
- Extended headers are copied correctly from first part
- The SSE detection logic is working correctly

Expected to show if chunks have SseType=0 (none) or proper SSE types set.

* s3: Trace SSE chunk metadata through multipart completion and retrieval

Added end-to-end logging to track SSE chunk metadata lifecycle:

**During Multipart Completion (filer_multipart.go)**:
1. Log finalParts chunks BEFORE mkFile - shows SseType and metadata
2. Log versionEntry.Chunks INSIDE mkFile callback - shows if mkFile preserves SSE info
3. Log success after mkFile completes

**During GET Retrieval (s3api_object_handlers.go)**:
1. Log retrieved entry chunks - shows SseType and metadata after retrieval
2. Log detected SSE type result

This will reveal at which point SSE chunk metadata is lost:
- If finalParts have SSE metadata but versionEntry.Chunks don't → mkFile bug
- If versionEntry.Chunks have SSE metadata but retrieved chunks don't → storage/retrieval bug
- If chunks never have SSE metadata → multipart completion SSE processing bug

Expected to show chunks with SseType=NONE during retrieval even though
they were created with proper SseType during multipart completion.

* s3: Fix SSE-C multipart IV base64 decoding bug

**Critical Bug Found**: SSE-C multipart uploads were failing because:

Root Cause:
- entry.Extended[SeaweedFSSSEIV] stores base64-encoded IV (24 bytes for 16-byte IV)
- SerializeSSECMetadata expects raw IV bytes (16 bytes)
- During multipart completion, we were passing base64 IV directly → serialization error

Error Message:
"Failed to serialize SSE-C metadata for chunk in part X: invalid IV length: expected 16 bytes, got 24"

Fix:
- Base64-decode IV before passing to SerializeSSECMetadata
- Added error handling for decode failures

Impact:
- SSE-C multipart uploads will now correctly serialize chunk metadata
- Chunks will have proper SSE metadata for decryption during GET

This fixes the SSE-C subtest of TestSSEMultipartUploadIntegration.
SSE-KMS still has a separate issue (error code 23) being investigated.

* fixes

* kms sse

* handle retry if not found in .versions folder and should read the normal object

* quick check (no retries) to see if the .versions/ directory exists

* skip retry if object is not found

* explicit update to avoid sync delay

* fix map update lock

* Remove fmt.Printf debug statements

* Fix SSE-KMS multipart base IV fallback to fail instead of regenerating

* fmt

* Fix ACL grants storage logic

* header handling

* nil handling

* range read for sse content

* test range requests for sse objects

* fmt

* unused code

* upload in chunks

* header case

* fix url

* bucket policy error vs bucket not found

* jwt handling

* fmt

* jwt in request header

* Optimize Case-Insensitive Prefix Check

* dead code

* Eliminated Unnecessary Stream Prefetch for Multipart SSE

* range sse

* sse

* refactor

* context

* fmt

* fix type

* fix SSE-C IV Mismatch

* Fix Headers Being Set After WriteHeader

* fix url parsing

* propergate sse headers

* multipart sse-s3

* aws sig v4 authen

* sse kms

* set content range

* better errors

* Update s3api_object_handlers_copy.go

* Update s3api_object_handlers.go

* Update s3api_object_handlers.go

* avoid magic number

* clean up

* Update s3api_bucket_policy_handlers.go

* fix url parsing

* context

* data and metadata both use background context

* adjust the offset

* SSE Range Request IV Calculation

* adjust logs

* IV relative to offset in each part, not the whole file

* collect logs

* offset

* fix offset

* fix url

* logs

* variable

* jwt

* Multipart ETag semantics: conditionally set object-level Md5 for single-chunk uploads only.

* sse

* adjust IV and offset

* multipart boundaries

* ensures PUT and GET operations return consistent ETags

* Metadata Header Case

* CommonPrefixes Sorting with URL Encoding

* always sort

* remove the extra PathUnescape call

* fix the multipart get part ETag

* the FileChunk is created without setting ModifiedTsNs

* Sort CommonPrefixes lexicographically to match AWS S3 behavior

* set md5 for multipart uploads

* prevents any potential data loss or corruption in the small-file inline storage path

* compiles correctly

* decryptedReader will now be properly closed after use

* Fixed URL encoding and sort order for CommonPrefixes

* Update s3api_object_handlers_list.go

* SSE-x Chunk View Decryption

* Different IV offset calculations for single-part vs multipart objects

* still too verbose in logs

* less logs

* ensure correct conversion

* fix listing

* nil check

* minor fixes

* nil check

* single character delimiter

* optimize

* range on empty object or zero-length

* correct IV based on its position within that part, not its position in the entire object

* adjust offset

* offset

Fetch FULL encrypted chunk (not just the range)
Adjust IV by PartOffset/ChunkOffset only
Decrypt full chunk
Skip in the DECRYPTED stream to reach OffsetInChunk

* look breaking

* refactor

* error on no content

* handle intra-block byte skipping

* Incomplete HTTP Response Error Handling

* multipart SSE

* Update s3api_object_handlers.go

* address comments

* less logs

* handling directory

* Optimized rejectDirectoryObjectWithoutSlash() to avoid unnecessary lookups

* Revert "handling directory"

This reverts commit 3a335f0ac33c63f51975abc63c40e5328857a74b.

* constant

* Consolidate nil entry checks in GetObjectHandler

* add range tests

* Consolidate redundant nil entry checks in HeadObjectHandler

* adjust logs

* SSE type

* large files

* large files

Reverted the plain-object range test

* ErrNoEncryptionConfig

* Fixed SSERangeReader Infinite Loop Vulnerability

* Fixed SSE-KMS Multipart ChunkReader HTTP Body Leak

* handle empty directory in S3, added PyArrow tests

* purge unused code

* Update s3_parquet_test.py

* Update requirements.txt

* According to S3 specifications, when both partNumber and Range are present, the Range should apply within the selected part's boundaries, not to the full object.

* handle errors

* errors after writing header

* https

* fix: Wait for volume assignment readiness before running Parquet tests

The test-implicit-dir-with-server test was failing with an Internal Error
because volume assignment was not ready when tests started. This fix adds
a check that attempts a volume assignment and waits for it to succeed
before proceeding with tests.

This ensures that:
1. Volume servers are registered with the master
2. Volume growth is triggered if needed
3. The system can successfully assign volumes for writes

Fixes the timeout issue where boto3 would retry 4 times and fail with
'We encountered an internal error, please try again.'

* sse tests

* store derived IV

* fix: Clean up gRPC ports between tests to prevent port conflicts

The second test (test-implicit-dir-with-server) was failing because the
volume server's gRPC port (18080 = VOLUME_PORT + 10000) was still in use
from the first test. The cleanup code only killed HTTP port processes,
not gRPC port processes.

Added cleanup for gRPC ports in all stop targets:
- Master gRPC: MASTER_PORT + 10000 (19333)
- Volume gRPC: VOLUME_PORT + 10000 (18080)
- Filer gRPC: FILER_PORT + 10000 (18888)

This ensures clean state between test runs in CI.

* add import

* address comments

* docs: Add placeholder documentation files for Parquet test suite

Added three missing documentation files referenced in test/s3/parquet/README.md:

1. TEST_COVERAGE.md - Documents 43 total test cases (17 Go unit tests,
   6 Python integration tests, 20 Python end-to-end tests)

2. FINAL_ROOT_CAUSE_ANALYSIS.md - Explains the s3fs compatibility issue
   with PyArrow, the implicit directory problem, and how the fix works

3. MINIO_DIRECTORY_HANDLING.md - Compares MinIO's directory handling
   approach with SeaweedFS's implementation

Each file contains:
- Title and overview
- Key technical details relevant to the topic
- TODO sections for future expansion

These placeholder files resolve the broken README links and provide
structure for future detailed documentation.

* clean up if metadata operation failed

* Update s3_parquet_test.py

* clean up

* Update Makefile

* Update s3_parquet_test.py

* Update Makefile

* Handle ivSkip for non-block-aligned offsets

* Update README.md

* stop volume server faster

* stop volume server in 1 second

* different IV for each chunk in SSE-S3 and SSE-KMS

* clean up if fails

* testing upload

* error propagation

* fmt

* simplify

* fix copying

* less logs

* endian

* Added marshaling error handling

* handling invalid ranges

* error handling for adding to log buffer

* fix logging

* avoid returning too quickly and ensure proper cleaning up

* Activity Tracking for Disk Reads

* Cleanup Unused Parameters

* Activity Tracking for Kafka Publishers

* Proper Test Error Reporting

* refactoring

* less logs

* less logs

* go fmt

* guard it with if entry.Attributes.TtlSec > 0 to match the pattern used elsewhere.

* Handle bucket-default encryption config errors explicitly for multipart

* consistent activity tracking

* obsolete code for s3 on filer read/write handlers

* Update weed/s3api/s3api_object_handlers_list.go

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

10 files changed, 437 insertions, 129 deletions

diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go
index 715dbdd30..22e69cc60 100644
--- a/weed/util/log_buffer/log_buffer.go
+++ b/weed/util/log_buffer/log_buffer.go
@@ -19,6 +19,12 @@ import (
 const BufferSize = 8 * 1024 * 1024
 const PreviousBufferCount = 32
 
+// Errors that can be returned by log buffer operations
+var (
+	// ErrBufferCorrupted indicates the log buffer contains corrupted data
+	ErrBufferCorrupted = fmt.Errorf("log buffer is corrupted")
+)
+
 type dataToFlush struct {
 	startTime time.Time
 	stopTime  time.Time
@@ -117,14 +123,12 @@ func (logBuffer *LogBuffer) RegisterSubscriber(subscriberID string) chan struct{
 
 	// Check if already registered
 	if existingChan, exists := logBuffer.subscribers[subscriberID]; exists {
-		glog.V(2).Infof("Subscriber %s already registered for %s, reusing channel", subscriberID, logBuffer.name)
 		return existingChan
 	}
 
 	// Create buffered channel (size 1) so notifications never block
 	notifyChan := make(chan struct{}, 1)
 	logBuffer.subscribers[subscriberID] = notifyChan
-	glog.V(1).Infof("Registered subscriber %s for %s (total: %d)", subscriberID, logBuffer.name, len(logBuffer.subscribers))
 	return notifyChan
 }
 
@@ -136,7 +140,6 @@ func (logBuffer *LogBuffer) UnregisterSubscriber(subscriberID string) {
 	if ch, exists := logBuffer.subscribers[subscriberID]; exists {
 		close(ch)
 		delete(logBuffer.subscribers, subscriberID)
-		glog.V(1).Infof("Unregistered subscriber %s from %s (remaining: %d)", subscriberID, logBuffer.name, len(logBuffer.subscribers))
 	}
 }
 
@@ -158,7 +161,6 @@ func (logBuffer *LogBuffer) IsOffsetInMemory(offset int64) bool {
 	// it MUST be in memory (not written to disk yet)
 	lastFlushed := logBuffer.lastFlushedOffset.Load()
 	if lastFlushed >= 0 && offset > lastFlushed {
-		glog.V(3).Infof("Offset %d is in memory (newer than lastFlushed=%d)", offset, lastFlushed)
 		return true
 	}
 
@@ -168,11 +170,9 @@ func (logBuffer *LogBuffer) IsOffsetInMemory(offset int64) bool {
 		// CRITICAL: Check if buffer actually has data (pos > 0)
 		// After flush, pos=0 but range is still valid - data is on disk, not in memory
 		if logBuffer.pos > 0 {
-			glog.V(3).Infof("Offset %d is in current buffer [%d-%d] with data", offset, logBuffer.bufferStartOffset, logBuffer.offset)
 			return true
 		}
 		// Buffer is empty (just flushed) - data is on disk
-		glog.V(3).Infof("Offset %d in range [%d-%d] but buffer empty (pos=0), data on disk", offset, logBuffer.bufferStartOffset, logBuffer.offset)
 		return false
 	}
 
@@ -181,17 +181,14 @@ func (logBuffer *LogBuffer) IsOffsetInMemory(offset int64) bool {
 		if offset >= buf.startOffset && offset <= buf.offset {
 			// Check if prevBuffer actually has data
 			if buf.size > 0 {
-				glog.V(3).Infof("Offset %d is in previous buffer [%d-%d] with data", offset, buf.startOffset, buf.offset)
 				return true
 			}
 			// Buffer is empty (flushed) - data is on disk
-			glog.V(3).Infof("Offset %d in prevBuffer [%d-%d] but empty (size=0), data on disk", offset, buf.startOffset, buf.offset)
 			return false
 		}
 	}
 
 	// Offset is older than memory buffers - only available on disk
-	glog.V(3).Infof("Offset %d is NOT in memory (bufferStart=%d, lastFlushed=%d)", offset, logBuffer.bufferStartOffset, lastFlushed)
 	return false
 }
 
@@ -205,15 +202,13 @@ func (logBuffer *LogBuffer) notifySubscribers() {
 		return // No subscribers, skip notification
 	}
 
-	for subscriberID, notifyChan := range logBuffer.subscribers {
+	for _, notifyChan := range logBuffer.subscribers {
 		select {
 		case notifyChan <- struct{}{}:
 			// Notification sent successfully
-			glog.V(3).Infof("Notified subscriber %s for %s", subscriberID, logBuffer.name)
 		default:
 			// Channel full - subscriber hasn't consumed previous notification yet
 			// This is OK because one notification is sufficient to wake the subscriber
-			glog.V(3).Infof("Subscriber %s notification channel full (OK - already notified)", subscriberID)
 		}
 	}
 }
@@ -227,7 +222,6 @@ func (logBuffer *LogBuffer) InitializeOffsetFromExistingData(getHighestOffsetFn
 
 	highestOffset, err := getHighestOffsetFn()
 	if err != nil {
-		glog.V(0).Infof("Failed to get highest offset for %s: %v, starting from 0", logBuffer.name, err)
 		return nil // Continue with offset 0 if we can't read existing data
 	}
 
@@ -243,37 +237,36 @@ func (logBuffer *LogBuffer) InitializeOffsetFromExistingData(getHighestOffsetFn
 		logBuffer.lastFlushedOffset.Store(highestOffset)
 		// Set lastFlushedTime to current time (we know data up to highestOffset is on disk)
 		logBuffer.lastFlushTsNs.Store(time.Now().UnixNano())
-		glog.V(0).Infof("Initialized LogBuffer %s offset to %d (highest existing: %d), buffer starts at %d, lastFlushedOffset=%d, lastFlushedTime=%v",
-			logBuffer.name, nextOffset, highestOffset, nextOffset, highestOffset, time.Now())
 	} else {
 		logBuffer.bufferStartOffset = 0 // Start from offset 0
 		// No data on disk yet
-		glog.V(0).Infof("No existing data found for %s, starting from offset 0, lastFlushedOffset=-1, lastFlushedTime=0", logBuffer.name)
 	}
 
 	return nil
 }
 
-func (logBuffer *LogBuffer) AddToBuffer(message *mq_pb.DataMessage) {
-	logBuffer.AddDataToBuffer(message.Key, message.Value, message.TsNs)
+func (logBuffer *LogBuffer) AddToBuffer(message *mq_pb.DataMessage) error {
+	return logBuffer.AddDataToBuffer(message.Key, message.Value, message.TsNs)
 }
 
 // AddLogEntryToBuffer directly adds a LogEntry to the buffer, preserving offset information
-func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) {
-	logEntryData, _ := proto.Marshal(logEntry)
-
+func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) error {
 	var toFlush *dataToFlush
+	var marshalErr error
 	logBuffer.Lock()
 	defer func() {
 		logBuffer.Unlock()
 		if toFlush != nil {
 			logBuffer.flushChan <- toFlush
 		}
-		if logBuffer.notifyFn != nil {
-			logBuffer.notifyFn()
+		// Only notify if there was no error
+		if marshalErr == nil {
+			if logBuffer.notifyFn != nil {
+				logBuffer.notifyFn()
+			}
+			// Notify all registered subscribers instantly (<1ms latency)
+			logBuffer.notifySubscribers()
 		}
-		// Notify all registered subscribers instantly (<1ms latency)
-		logBuffer.notifySubscribers()
 	}()
 
 	processingTsNs := logEntry.TsNs
@@ -285,11 +278,16 @@ func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) {
 		ts = time.Unix(0, processingTsNs)
 		// Re-marshal with corrected timestamp
 		logEntry.TsNs = processingTsNs
-		logEntryData, _ = proto.Marshal(logEntry)
 	} else {
 		logBuffer.LastTsNs.Store(processingTsNs)
 	}
 
+	logEntryData, err := proto.Marshal(logEntry)
+	if err != nil {
+		marshalErr = fmt.Errorf("failed to marshal LogEntry: %w", err)
+		glog.Errorf("%v", marshalErr)
+		return marshalErr
+	}
 	size := len(logEntryData)
 
 	if logBuffer.pos == 0 {
@@ -323,8 +321,9 @@ func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) {
 			const maxBufferSize = 1 << 30 // 1 GiB practical limit
 			// Ensure 2*size + 4 won't overflow int and stays within practical bounds
 			if size < 0 || size > (math.MaxInt-4)/2 || size > (maxBufferSize-4)/2 {
-				glog.Errorf("Buffer size out of valid range: %d bytes, skipping", size)
-				return
+				marshalErr = fmt.Errorf("message size %d exceeds maximum allowed size", size)
+				glog.Errorf("%v", marshalErr)
+				return marshalErr
 			}
 			// Safe to compute now that we've validated size is in valid range
 			newSize := 2*size + 4
@@ -340,9 +339,10 @@ func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) {
 	logBuffer.pos += size + 4
 
 	logBuffer.offset++
+	return nil
 }
 
-func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processingTsNs int64) {
+func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processingTsNs int64) error {
 
 	// PERFORMANCE OPTIMIZATION: Pre-process expensive operations OUTSIDE the lock
 	var ts time.Time
@@ -360,20 +360,22 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 		Key:              partitionKey,
 	}
 
-	logEntryData, _ := proto.Marshal(logEntry)
-
 	var toFlush *dataToFlush
+	var marshalErr error
 	logBuffer.Lock()
 	defer func() {
 		logBuffer.Unlock()
 		if toFlush != nil {
 			logBuffer.flushChan <- toFlush
 		}
-		if logBuffer.notifyFn != nil {
-			logBuffer.notifyFn()
+		// Only notify if there was no error
+		if marshalErr == nil {
+			if logBuffer.notifyFn != nil {
+				logBuffer.notifyFn()
+			}
+			// Notify all registered subscribers instantly (<1ms latency)
+			logBuffer.notifySubscribers()
 		}
-		// Notify all registered subscribers instantly (<1ms latency)
-		logBuffer.notifySubscribers()
 	}()
 
 	// Handle timestamp collision inside lock (rare case)
@@ -390,20 +392,13 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	// Note: This also enables AddToBuffer to work correctly with Kafka-style offset-based reads
 	logEntry.Offset = logBuffer.offset
 
-	// DEBUG: Log data being added to buffer for GitHub Actions debugging
-	dataPreview := ""
-	if len(data) > 0 {
-		if len(data) <= 50 {
-			dataPreview = string(data)
-		} else {
-			dataPreview = fmt.Sprintf("%s...(total %d bytes)", string(data[:50]), len(data))
-		}
-	}
-	glog.V(2).Infof("[LOG_BUFFER_ADD] buffer=%s offset=%d dataLen=%d dataPreview=%q",
-		logBuffer.name, logBuffer.offset, len(data), dataPreview)
-
 	// Marshal with correct timestamp and offset
-	logEntryData, _ = proto.Marshal(logEntry)
+	logEntryData, err := proto.Marshal(logEntry)
+	if err != nil {
+		marshalErr = fmt.Errorf("failed to marshal LogEntry: %w", err)
+		glog.Errorf("%v", marshalErr)
+		return marshalErr
+	}
 
 	size := len(logEntryData)
 
@@ -429,7 +424,6 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	}
 
 	if logBuffer.startTime.Add(logBuffer.flushInterval).Before(ts) || len(logBuffer.buf)-logBuffer.pos < size+4 {
-		// glog.V(0).Infof("%s copyToFlush1 offset:%d count:%d start time %v, ts %v, remaining %d bytes", logBuffer.name, logBuffer.offset, len(logBuffer.idx), logBuffer.startTime, ts, len(logBuffer.buf)-logBuffer.pos)
 		toFlush = logBuffer.copyToFlush()
 		logBuffer.startTime = ts
 		if len(logBuffer.buf) < size+4 {
@@ -437,8 +431,9 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 			const maxBufferSize = 1 << 30 // 1 GiB practical limit
 			// Ensure 2*size + 4 won't overflow int and stays within practical bounds
 			if size < 0 || size > (math.MaxInt-4)/2 || size > (maxBufferSize-4)/2 {
-				glog.Errorf("Buffer size out of valid range: %d bytes, skipping", size)
-				return
+				marshalErr = fmt.Errorf("message size %d exceeds maximum allowed size", size)
+				glog.Errorf("%v", marshalErr)
+				return marshalErr
 			}
 			// Safe to compute now that we've validated size is in valid range
 			newSize := 2*size + 4
@@ -454,6 +449,7 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	logBuffer.pos += size + 4
 
 	logBuffer.offset++
+	return nil
 }
 
 func (logBuffer *LogBuffer) IsStopping() bool {
@@ -480,14 +476,11 @@ func (logBuffer *LogBuffer) ForceFlush() {
 			select {
 			case <-toFlush.done:
 				// Flush completed successfully
-				glog.V(1).Infof("ForceFlush completed for %s", logBuffer.name)
 			case <-time.After(5 * time.Second):
 				// Timeout waiting for flush - this shouldn't happen
-				glog.Warningf("ForceFlush timed out waiting for completion on %s", logBuffer.name)
 			}
 		case <-time.After(2 * time.Second):
 			// If flush channel is still blocked after 2s, something is wrong
-			glog.Warningf("ForceFlush channel timeout for %s - flush channel busy for 2s", logBuffer.name)
 		}
 	}
 }
@@ -511,7 +504,6 @@ func (logBuffer *LogBuffer) IsAllFlushed() bool {
 func (logBuffer *LogBuffer) loopFlush() {
 	for d := range logBuffer.flushChan {
 		if d != nil {
-			// glog.V(4).Infof("%s flush [%v, %v] size %d", m.name, d.startTime, d.stopTime, len(d.data.Bytes()))
 			logBuffer.flushFn(logBuffer, d.startTime, d.stopTime, d.data.Bytes(), d.minOffset, d.maxOffset)
 			d.releaseMemory()
 			// local logbuffer is different from aggregate logbuffer here
@@ -546,10 +538,7 @@ func (logBuffer *LogBuffer) loopInterval() {
 		toFlush := logBuffer.copyToFlush()
 		logBuffer.Unlock()
 		if toFlush != nil {
-			glog.V(4).Infof("%s flush [%v, %v] size %d", logBuffer.name, toFlush.startTime, toFlush.stopTime, len(toFlush.data.Bytes()))
 			logBuffer.flushChan <- toFlush
-		} else {
-			// glog.V(0).Infof("%s no flush", m.name)
 		}
 	}
 }
@@ -578,9 +567,7 @@ func (logBuffer *LogBuffer) copyToFlushInternal(withCallback bool) *dataToFlush
 			if withCallback {
 				d.done = make(chan struct{})
 			}
-			// glog.V(4).Infof("%s flushing [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime)
 		} else {
-			// glog.V(4).Infof("%s removed from memory [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime)
 			logBuffer.lastFlushDataTime = logBuffer.stopTime
 		}
 		// CRITICAL: logBuffer.offset is the "next offset to assign", so last offset in buffer is offset-1
@@ -647,8 +634,6 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 	defer logBuffer.RUnlock()
 
 	isOffsetBased := lastReadPosition.IsOffsetBased
-	glog.V(2).Infof("[ReadFromBuffer] %s: isOffsetBased=%v, position=%+v, bufferStartOffset=%d, offset=%d, pos=%d",
-		logBuffer.name, isOffsetBased, lastReadPosition, logBuffer.bufferStartOffset, logBuffer.offset, logBuffer.pos)
 
 	// For offset-based subscriptions, use offset comparisons, not time comparisons!
 	if isOffsetBased {
@@ -729,11 +714,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 	if !logBuffer.startTime.IsZero() {
 		tsMemory = logBuffer.startTime
 	}
-	glog.V(2).Infof("[ReadFromBuffer] %s: checking prevBuffers, count=%d, currentStartTime=%v",
-		logBuffer.name, len(logBuffer.prevBuffers.buffers), logBuffer.startTime)
-	for i, prevBuf := range logBuffer.prevBuffers.buffers {
-		glog.V(2).Infof("[ReadFromBuffer] %s: prevBuf[%d]: startTime=%v stopTime=%v size=%d startOffset=%d endOffset=%d",
-			logBuffer.name, i, prevBuf.startTime, prevBuf.stopTime, prevBuf.size, prevBuf.startOffset, prevBuf.offset)
+	for _, prevBuf := range logBuffer.prevBuffers.buffers {
 		if !prevBuf.startTime.IsZero() {
 			// If tsMemory is zero, assign directly; otherwise compare
 			if tsMemory.IsZero() || prevBuf.startTime.Before(tsMemory) {
@@ -754,19 +735,12 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 			// Fall through to case 2.1 to read from earliest buffer
 		} else if lastReadPosition.Offset <= 0 && lastReadPosition.Time.Before(tsMemory) {
 			// Treat first read with sentinel/zero offset as inclusive of earliest in-memory data
-			glog.V(4).Infof("first read (offset=%d) at time %v before earliest memory %v, reading from memory",
-				lastReadPosition.Offset, lastReadPosition.Time, tsMemory)
 		} else {
 			// Data not in memory buffers - read from disk
-			glog.V(0).Infof("[ReadFromBuffer] %s resume from disk: requested time %v < earliest memory time %v",
-				logBuffer.name, lastReadPosition.Time, tsMemory)
 			return nil, -2, ResumeFromDiskError
 		}
 	}
 
-	glog.V(2).Infof("[ReadFromBuffer] %s: time-based read continuing, tsMemory=%v, lastReadPos=%v",
-		logBuffer.name, tsMemory, lastReadPosition.Time)
-
 	// the following is case 2.1
 
 	if lastReadPosition.Time.Equal(logBuffer.stopTime) && !logBuffer.stopTime.IsZero() {
@@ -776,14 +750,12 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		}
 	}
 	if lastReadPosition.Time.After(logBuffer.stopTime) && !logBuffer.stopTime.IsZero() {
-		// glog.Fatalf("unexpected last read time %v, older than latest %v", lastReadPosition, m.stopTime)
 		return nil, logBuffer.offset, nil
 	}
 	// Also check prevBuffers when current buffer is empty (startTime is zero)
 	if lastReadPosition.Time.Before(logBuffer.startTime) || logBuffer.startTime.IsZero() {
 		for _, buf := range logBuffer.prevBuffers.buffers {
 			if buf.startTime.After(lastReadPosition.Time) {
-				// glog.V(4).Infof("%s return the %d sealed buffer %v", m.name, i, buf.startTime)
 				return copiedBytes(buf.buf[:buf.size]), buf.offset, nil
 			}
 			if !buf.startTime.After(lastReadPosition.Time) && buf.stopTime.After(lastReadPosition.Time) {
@@ -791,14 +763,17 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 				if lastReadPosition.Offset <= 0 {
 					searchTime = searchTime.Add(-time.Nanosecond)
 				}
-				pos := buf.locateByTs(searchTime)
-				glog.V(2).Infof("[ReadFromBuffer] %s: found data in prevBuffer at pos %d, bufSize=%d", logBuffer.name, pos, buf.size)
+				pos, err := buf.locateByTs(searchTime)
+				if err != nil {
+					// Buffer corruption detected - return error wrapped with ErrBufferCorrupted
+					glog.Errorf("ReadFromBuffer: buffer corruption in prevBuffer: %v", err)
+					return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
+				}
 				return copiedBytes(buf.buf[pos:buf.size]), buf.offset, nil
 			}
 		}
 		// If current buffer is not empty, return it
 		if logBuffer.pos > 0 {
-			// glog.V(4).Infof("%s return the current buf %v", m.name, lastReadPosition)
 			return copiedBytes(logBuffer.buf[:logBuffer.pos]), logBuffer.offset, nil
 		}
 		// Buffer is empty and no data in prevBuffers - wait for new data
@@ -830,13 +805,23 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 	for l <= h {
 		mid := (l + h) / 2
 		pos := logBuffer.idx[mid]
-		_, t := readTs(logBuffer.buf, pos)
+		_, t, err := readTs(logBuffer.buf, pos)
+		if err != nil {
+			// Buffer corruption detected in binary search
+			glog.Errorf("ReadFromBuffer: buffer corruption at idx[%d] pos %d: %v", mid, pos, err)
+			return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
+		}
 		if t <= searchTs {
 			l = mid + 1
 		} else if searchTs < t {
 			var prevT int64
 			if mid > 0 {
-				_, prevT = readTs(logBuffer.buf, logBuffer.idx[mid-1])
+				_, prevT, err = readTs(logBuffer.buf, logBuffer.idx[mid-1])
+				if err != nil {
+					// Buffer corruption detected in binary search (previous entry)
+					glog.Errorf("ReadFromBuffer: buffer corruption at idx[%d] pos %d: %v", mid-1, logBuffer.idx[mid-1], err)
+					return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
+				}
 			}
 			if prevT <= searchTs {
 				return copiedBytes(logBuffer.buf[pos:logBuffer.pos]), logBuffer.offset, nil
@@ -881,16 +866,28 @@ func copiedBytes(buf []byte) (copied *bytes.Buffer) {
 	return
 }
 
-func readTs(buf []byte, pos int) (size int, ts int64) {
+func readTs(buf []byte, pos int) (size int, ts int64, err error) {
+	// Bounds check for size field (overflow-safe)
+	if pos < 0 || pos > len(buf)-4 {
+		return 0, 0, fmt.Errorf("corrupted log buffer: cannot read size at pos %d, buffer length %d", pos, len(buf))
+	}
 
 	size = int(util.BytesToUint32(buf[pos : pos+4]))
+
+	// Bounds check for entry data (overflow-safe, protects against negative size)
+	if size < 0 || size > len(buf)-pos-4 {
+		return 0, 0, fmt.Errorf("corrupted log buffer: entry size %d at pos %d exceeds buffer length %d", size, pos, len(buf))
+	}
+
 	entryData := buf[pos+4 : pos+4+size]
 	logEntry := &filer_pb.LogEntry{}
 
-	err := proto.Unmarshal(entryData, logEntry)
+	err = proto.Unmarshal(entryData, logEntry)
 	if err != nil {
-		glog.Fatalf("unexpected unmarshal filer_pb.LogEntry: %v", err)
+		// Return error instead of failing fast
+		// This allows caller to handle corruption gracefully
+		return 0, 0, fmt.Errorf("corrupted log buffer: failed to unmarshal LogEntry at pos %d, size %d: %w", pos, size, err)
 	}
-	return size, logEntry.TsNs
+	return size, logEntry.TsNs, nil
 
 }
diff --git a/weed/util/log_buffer/log_buffer_corruption_test.go b/weed/util/log_buffer/log_buffer_corruption_test.go
new file mode 100644
index 000000000..2f7a029e6
--- /dev/null
+++ b/weed/util/log_buffer/log_buffer_corruption_test.go
@@ -0,0 +1,224 @@
+package log_buffer
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// TestReadTsCorruptedBuffer tests that readTs properly returns an error for corrupted data
+func TestReadTsCorruptedBuffer(t *testing.T) {
+	// Create a corrupted buffer with invalid protobuf data
+	buf := make([]byte, 100)
+
+	// Set size field to 10 bytes (using proper encoding)
+	util.Uint32toBytes(buf[0:4], 10)
+
+	// Fill with garbage data that won't unmarshal as LogEntry
+	for i := 4; i < 14; i++ {
+		buf[i] = 0xFF
+	}
+
+	// Attempt to read timestamp
+	size, ts, err := readTs(buf, 0)
+
+	// Should return an error
+	if err == nil {
+		t.Error("Expected error for corrupted buffer, got nil")
+	}
+
+	// Size and ts should be zero on error
+	if size != 0 {
+		t.Errorf("Expected size=0 on error, got %d", size)
+	}
+
+	if ts != 0 {
+		t.Errorf("Expected ts=0 on error, got %d", ts)
+	}
+
+	// Error should indicate corruption
+	if !errors.Is(err, ErrBufferCorrupted) {
+		t.Logf("Error message: %v", err)
+		// Check if error message contains expected text
+		if err.Error() == "" || len(err.Error()) == 0 {
+			t.Error("Expected non-empty error message")
+		}
+	}
+
+	t.Logf("✓ readTs correctly returned error for corrupted buffer: %v", err)
+}
+
+// TestReadTsValidBuffer tests that readTs works correctly for valid data
+func TestReadTsValidBuffer(t *testing.T) {
+	// Create a valid LogEntry
+	logEntry := &filer_pb.LogEntry{
+		TsNs: 123456789,
+		Key:  []byte("test-key"),
+	}
+
+	// Marshal it
+	data, err := proto.Marshal(logEntry)
+	if err != nil {
+		t.Fatalf("Failed to marshal LogEntry: %v", err)
+	}
+
+	// Create buffer with size prefix using util function
+	buf := make([]byte, 4+len(data))
+	util.Uint32toBytes(buf[0:4], uint32(len(data)))
+	copy(buf[4:], data)
+
+	// Read timestamp
+	size, ts, err := readTs(buf, 0)
+
+	// Should succeed
+	if err != nil {
+		t.Fatalf("Expected no error for valid buffer, got: %v", err)
+	}
+
+	// Should return correct values
+	if size != len(data) {
+		t.Errorf("Expected size=%d, got %d", len(data), size)
+	}
+
+	if ts != logEntry.TsNs {
+		t.Errorf("Expected ts=%d, got %d", logEntry.TsNs, ts)
+	}
+
+	t.Logf("✓ readTs correctly parsed valid buffer: size=%d, ts=%d", size, ts)
+}
+
+// TestReadFromBufferCorruption tests that ReadFromBuffer propagates corruption errors
+func TestReadFromBufferCorruption(t *testing.T) {
+	lb := NewLogBuffer("test-corruption", time.Second, nil, nil, func() {})
+
+	// Add a valid entry first using AddDataToBuffer
+	validKey := []byte("valid")
+	validData, _ := proto.Marshal(&filer_pb.LogEntry{
+		TsNs: 1000,
+		Key:  validKey,
+	})
+	if err := lb.AddDataToBuffer(validKey, validData, 1000); err != nil {
+		t.Fatalf("Failed to add data to buffer: %v", err)
+	}
+
+	// Manually corrupt the buffer by writing garbage
+	// This simulates a corruption scenario
+	if len(lb.idx) > 0 {
+		pos := lb.idx[0]
+		// Overwrite the protobuf data with garbage
+		for i := pos + 4; i < pos+8 && i < len(lb.buf); i++ {
+			lb.buf[i] = 0xFF
+		}
+	}
+
+	// Try to read - should detect corruption
+	startPos := MessagePosition{Time: lb.startTime}
+	buf, offset, err := lb.ReadFromBuffer(startPos)
+
+	// Should return corruption error
+	if err == nil {
+		t.Error("Expected corruption error, got nil")
+		if buf != nil {
+			t.Logf("Unexpected success: got buffer with %d bytes", buf.Len())
+		}
+	} else {
+		// Verify it's a corruption error
+		if !errors.Is(err, ErrBufferCorrupted) {
+			t.Logf("Got error (not ErrBufferCorrupted sentinel, but still an error): %v", err)
+		}
+		t.Logf("✓ ReadFromBuffer correctly detected corruption: %v", err)
+	}
+
+	t.Logf("ReadFromBuffer result: buf=%v, offset=%d, err=%v", buf != nil, offset, err)
+}
+
+// TestLocateByTsCorruption tests that locateByTs propagates corruption errors
+func TestLocateByTsCorruption(t *testing.T) {
+	// Create a MemBuffer with corrupted data
+	mb := &MemBuffer{
+		buf:  make([]byte, 100),
+		size: 14,
+	}
+
+	// Set size field (using proper encoding)
+	util.Uint32toBytes(mb.buf[0:4], 10)
+
+	// Fill with garbage
+	for i := 4; i < 14; i++ {
+		mb.buf[i] = 0xFF
+	}
+
+	// Try to locate by timestamp
+	pos, err := mb.locateByTs(mb.startTime)
+
+	// Should return error
+	if err == nil {
+		t.Errorf("Expected corruption error, got nil (pos=%d)", pos)
+	} else {
+		t.Logf("✓ locateByTs correctly detected corruption: %v", err)
+	}
+}
+
+// TestErrorPropagationChain tests the complete error propagation from readTs -> locateByTs -> ReadFromBuffer
+func TestErrorPropagationChain(t *testing.T) {
+	t.Run("Corruption in readTs", func(t *testing.T) {
+		// Already covered by TestReadTsCorruptedBuffer
+		t.Log("✓ readTs error propagation tested")
+	})
+
+	t.Run("Corruption in locateByTs", func(t *testing.T) {
+		// Already covered by TestLocateByTsCorruption
+		t.Log("✓ locateByTs error propagation tested")
+	})
+
+	t.Run("Corruption in ReadFromBuffer binary search", func(t *testing.T) {
+		// Already covered by TestReadFromBufferCorruption
+		t.Log("✓ ReadFromBuffer error propagation tested")
+	})
+
+	t.Log("✓ Complete error propagation chain verified")
+}
+
+// TestNoSilentCorruption verifies that corruption never returns (0, 0) silently
+func TestNoSilentCorruption(t *testing.T) {
+	// Create various corrupted buffers
+	testCases := []struct {
+		name string
+		buf  []byte
+		pos  int
+	}{
+		{
+			name: "Invalid protobuf",
+			buf:  []byte{10, 0, 0, 0, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF},
+			pos:  0,
+		},
+		{
+			name: "Truncated data",
+			buf:  []byte{100, 0, 0, 0, 1, 2, 3}, // Size says 100 but only 3 bytes available
+			pos:  0,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			size, ts, err := readTs(tc.buf, tc.pos)
+
+			// CRITICAL: Must return error, never silent (0, 0)
+			if err == nil {
+				t.Errorf("CRITICAL: readTs returned (%d, %d, nil) for corrupted buffer - this causes silent data corruption!", size, ts)
+			} else {
+				t.Logf("✓ Correctly returned error instead of silent (0, 0): %v", err)
+			}
+
+			// On error, size and ts should be 0
+			if size != 0 || ts != 0 {
+				t.Errorf("On error, expected (0, 0), got (%d, %d)", size, ts)
+			}
+		})
+	}
+}
diff --git a/weed/util/log_buffer/log_buffer_flush_gap_test.go b/weed/util/log_buffer/log_buffer_flush_gap_test.go
index bc40ea6df..dc010f1b8 100644
--- a/weed/util/log_buffer/log_buffer_flush_gap_test.go
+++ b/weed/util/log_buffer/log_buffer_flush_gap_test.go
@@ -69,11 +69,13 @@ func TestFlushOffsetGap_ReproduceDataLoss(t *testing.T) {
 	t.Logf("Sending %d messages...", messageCount)
 
 	for i := 0; i < messageCount; i++ {
-		logBuffer.AddToBuffer(&mq_pb.DataMessage{
+		if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
 			Key:   []byte(fmt.Sprintf("key-%d", i)),
 			Value: []byte(fmt.Sprintf("message-%d", i)),
 			TsNs:  time.Now().UnixNano(),
-		})
+		}); err != nil {
+			t.Fatalf("Failed to add buffer: %v", err)
+		}
 	}
 
 	// Force flush multiple times to simulate real workload
@@ -82,11 +84,13 @@ func TestFlushOffsetGap_ReproduceDataLoss(t *testing.T) {
 
 	// Add more messages after flush
 	for i := messageCount; i < messageCount+50; i++ {
-		logBuffer.AddToBuffer(&mq_pb.DataMessage{
+		if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
 			Key:   []byte(fmt.Sprintf("key-%d", i)),
 			Value: []byte(fmt.Sprintf("message-%d", i)),
 			TsNs:  time.Now().UnixNano(),
-		})
+		}); err != nil {
+			t.Fatalf("Failed to add buffer: %v", err)
+		}
 	}
 
 	// Force another flush
@@ -209,11 +213,13 @@ func TestFlushOffsetGap_CheckPrevBuffers(t *testing.T) {
 		// Send 20 messages
 		for i := 0; i < 20; i++ {
 			offset := int64(batch*20 + i)
-			logBuffer.AddToBuffer(&mq_pb.DataMessage{
+			if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
 				Key:   []byte(fmt.Sprintf("key-%d", offset)),
 				Value: []byte(fmt.Sprintf("message-%d", offset)),
 				TsNs:  time.Now().UnixNano(),
-			})
+			}); err != nil {
+				t.Fatalf("Failed to add buffer: %v", err)
+			}
 		}
 
 		// Check state before flush
@@ -285,11 +291,14 @@ func TestFlushOffsetGap_ConcurrentWriteAndFlush(t *testing.T) {
 	go func() {
 		defer wg.Done()
 		for i := 0; i < 200; i++ {
-			logBuffer.AddToBuffer(&mq_pb.DataMessage{
+			if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
 				Key:   []byte(fmt.Sprintf("key-%d", i)),
 				Value: []byte(fmt.Sprintf("message-%d", i)),
 				TsNs:  time.Now().UnixNano(),
-			})
+			}); err != nil {
+				t.Errorf("Failed to add buffer: %v", err)
+				return
+			}
 			if i%50 == 0 {
 				time.Sleep(10 * time.Millisecond)
 			}
@@ -389,7 +398,9 @@ func TestFlushOffsetGap_ProductionScenario(t *testing.T) {
 			TsNs:   time.Now().UnixNano(),
 			Offset: nextKafkaOffset, // Explicit Kafka offset
 		}
-		logBuffer.AddLogEntryToBuffer(logEntry)
+		if err := logBuffer.AddLogEntryToBuffer(logEntry); err != nil {
+			t.Fatalf("Failed to add log entry: %v", err)
+		}
 		nextKafkaOffset++
 	}
 
@@ -422,7 +433,9 @@ func TestFlushOffsetGap_ProductionScenario(t *testing.T) {
 			TsNs:   time.Now().UnixNano(),
 			Offset: nextKafkaOffset,
 		}
-		logBuffer.AddLogEntryToBuffer(logEntry)
+		if err := logBuffer.AddLogEntryToBuffer(logEntry); err != nil {
+			t.Fatalf("Failed to add log entry: %v", err)
+		}
 		nextKafkaOffset++
 	}
 
@@ -546,7 +559,9 @@ func TestFlushOffsetGap_ConcurrentReadDuringFlush(t *testing.T) {
 			TsNs:   time.Now().UnixNano(),
 			Offset: i,
 		}
-		logBuffer.AddLogEntryToBuffer(logEntry)
+		if err := logBuffer.AddLogEntryToBuffer(logEntry); err != nil {
+			t.Fatalf("Failed to add log entry: %v", err)
+		}
 	}
 
 	// Flush (moves data to disk)
@@ -616,11 +631,13 @@ func TestFlushOffsetGap_ForceFlushAdvancesBuffer(t *testing.T) {
 
 		// Add 10 messages
 		for i := 0; i < 10; i++ {
-			logBuffer.AddToBuffer(&mq_pb.DataMessage{
+			if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
 				Key:   []byte(fmt.Sprintf("round-%d-msg-%d", round, i)),
 				Value: []byte(fmt.Sprintf("data-%d-%d", round, i)),
 				TsNs:  time.Now().UnixNano(),
-			})
+			}); err != nil {
+				t.Fatalf("Failed to add buffer: %v", err)
+			}
 		}
 
 		// Check state after adding
diff --git a/weed/util/log_buffer/log_buffer_queryability_test.go b/weed/util/log_buffer/log_buffer_queryability_test.go
index 16dd0f9b0..4774f25d8 100644
--- a/weed/util/log_buffer/log_buffer_queryability_test.go
+++ b/weed/util/log_buffer/log_buffer_queryability_test.go
@@ -39,7 +39,9 @@ func TestBufferQueryability(t *testing.T) {
 	}
 
 	// Add the entry to the buffer
-	logBuffer.AddLogEntryToBuffer(logEntry)
+	if err := logBuffer.AddLogEntryToBuffer(logEntry); err != nil {
+		t.Fatalf("Failed to add log entry: %v", err)
+	}
 
 	// Verify the buffer has data
 	if logBuffer.pos == 0 {
@@ -122,7 +124,9 @@ func TestMultipleEntriesQueryability(t *testing.T) {
 			Key:              []byte("test-key-" + string(rune('0'+i))),
 			Offset:           int64(i),
 		}
-		logBuffer.AddLogEntryToBuffer(logEntry)
+		if err := logBuffer.AddLogEntryToBuffer(logEntry); err != nil {
+			t.Fatalf("Failed to add log entry: %v", err)
+		}
 	}
 
 	// Read all entries
@@ -197,7 +201,9 @@ func TestSchemaRegistryScenario(t *testing.T) {
 	}
 
 	// Add to buffer
-	logBuffer.AddLogEntryToBuffer(logEntry)
+	if err := logBuffer.AddLogEntryToBuffer(logEntry); err != nil {
+		t.Fatalf("Failed to add log entry: %v", err)
+	}
 
 	// Simulate the SQL query scenario - read from offset 0
 	startPosition := NewMessagePosition(0, 0)
@@ -255,7 +261,9 @@ func TestTimeBasedFirstReadBeforeEarliest(t *testing.T) {
 	// Seed one entry so earliestTime is set
 	baseTs := time.Now().Add(-time.Second)
 	entry := &filer_pb.LogEntry{TsNs: baseTs.UnixNano(), Data: []byte("x"), Key: []byte("k"), Offset: 0}
-	logBuffer.AddLogEntryToBuffer(entry)
+	if err := logBuffer.AddLogEntryToBuffer(entry); err != nil {
+		t.Fatalf("Failed to add log entry: %v", err)
+	}
 	_ = flushed
 
 	// Start read 1ns before earliest memory, with offset sentinel (-2)
@@ -280,7 +288,9 @@ func TestEarliestTimeExactRead(t *testing.T) {
 
 	ts := time.Now()
 	entry := &filer_pb.LogEntry{TsNs: ts.UnixNano(), Data: []byte("a"), Key: []byte("k"), Offset: 0}
-	logBuffer.AddLogEntryToBuffer(entry)
+	if err := logBuffer.AddLogEntryToBuffer(entry); err != nil {
+		t.Fatalf("Failed to add log entry: %v", err)
+	}
 
 	startPos := NewMessagePosition(ts.UnixNano(), -2)
 	buf, _, err := logBuffer.ReadFromBuffer(startPos)
diff --git a/weed/util/log_buffer/log_buffer_test.go b/weed/util/log_buffer/log_buffer_test.go
index 7b851de06..d99a8f20c 100644
--- a/weed/util/log_buffer/log_buffer_test.go
+++ b/weed/util/log_buffer/log_buffer_test.go
@@ -52,11 +52,13 @@ func TestNewLogBufferFirstBuffer(t *testing.T) {
 	var buf = make([]byte, messageSize)
 	for i := 0; i < messageCount; i++ {
 		rand.Read(buf)
-		lb.AddToBuffer(&mq_pb.DataMessage{
+		if err := lb.AddToBuffer(&mq_pb.DataMessage{
 			Key:   nil,
 			Value: buf,
 			TsNs:  0,
-		})
+		}); err != nil {
+			t.Fatalf("Failed to add buffer: %v", err)
+		}
 	}
 	wg.Wait()
 
@@ -141,12 +143,14 @@ func TestReadFromBuffer_OldOffsetReturnsResumeFromDiskError(t *testing.T) {
 			if tt.hasData {
 				testData := []byte("test message")
 				// Use AddLogEntryToBuffer to preserve offset information
-				lb.AddLogEntryToBuffer(&filer_pb.LogEntry{
+				if err := lb.AddLogEntryToBuffer(&filer_pb.LogEntry{
 					TsNs:   time.Now().UnixNano(),
 					Key:    []byte("key"),
 					Data:   testData,
 					Offset: tt.currentOffset, // Add data at current offset
-				})
+				}); err != nil {
+					t.Fatalf("Failed to add log entry: %v", err)
+				}
 			}
 
 			// Create an offset-based position for the requested offset
@@ -365,11 +369,13 @@ func TestReadFromBuffer_InitializedFromDisk(t *testing.T) {
 		lb.offset, lb.bufferStartOffset)
 
 	// Now write a new message at offset 4
-	lb.AddToBuffer(&mq_pb.DataMessage{
+	if err := lb.AddToBuffer(&mq_pb.DataMessage{
 		Key:   []byte("new-key"),
 		Value: []byte("new-message-at-offset-4"),
 		TsNs:  time.Now().UnixNano(),
-	})
+	}); err != nil {
+		t.Fatalf("Failed to add buffer: %v", err)
+	}
 	// After AddToBuffer: offset=5, pos>0
 
 	// Schema Registry tries to read offset 0 (should be on disk)
@@ -503,11 +509,13 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) {
 
 	// Now add data and flush it
 	t.Logf("➕ Adding message to buffer...")
-	logBuffer.AddToBuffer(&mq_pb.DataMessage{
+	if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
 		Key:   []byte("key-0"),
 		Value: []byte("message-0"),
 		TsNs:  time.Now().UnixNano(),
-	})
+	}); err != nil {
+		t.Fatalf("Failed to add buffer: %v", err)
+	}
 
 	// Force flush
 	t.Logf("Force flushing...")
diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go
index 950604022..0a2b8e89a 100644
--- a/weed/util/log_buffer/log_read.go
+++ b/weed/util/log_buffer/log_read.go
@@ -2,6 +2,7 @@ package log_buffer
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"time"
 
@@ -77,6 +78,16 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 			time.Sleep(1127 * time.Millisecond)
 			return lastReadPosition, isDone, ResumeFromDiskError
 		}
+		if err != nil {
+			// Check for buffer corruption error
+			if errors.Is(err, ErrBufferCorrupted) {
+				glog.Errorf("%s: Buffer corruption detected: %v", readerName, err)
+				return lastReadPosition, true, fmt.Errorf("buffer corruption: %w", err)
+			}
+			// Other errors
+			glog.Errorf("%s: ReadFromBuffer error: %v", readerName, err)
+			return lastReadPosition, true, err
+		}
 		readSize := 0
 		if bytesBuf != nil {
 			readSize = bytesBuf.Len()
@@ -212,6 +223,13 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 		}
 		bytesBuf, offset, err = logBuffer.ReadFromBuffer(lastReadPosition)
 		glog.V(4).Infof("ReadFromBuffer for %s returned bytesBuf=%v, offset=%d, err=%v", readerName, bytesBuf != nil, offset, err)
+
+		// Check for buffer corruption error before other error handling
+		if err != nil && errors.Is(err, ErrBufferCorrupted) {
+			glog.Errorf("%s: Buffer corruption detected: %v", readerName, err)
+			return lastReadPosition, true, fmt.Errorf("buffer corruption: %w", err)
+		}
+
 		if err == ResumeFromDiskError {
 			// Try to read from disk if readFromDiskFn is available
 			if logBuffer.ReadFromDiskFn != nil {
diff --git a/weed/util/log_buffer/log_read_integration_test.go b/weed/util/log_buffer/log_read_integration_test.go
index 38549b9f7..8970ca683 100644
--- a/weed/util/log_buffer/log_read_integration_test.go
+++ b/weed/util/log_buffer/log_read_integration_test.go
@@ -31,7 +31,10 @@ func TestConcurrentProducerConsumer(t *testing.T) {
 				Data:   []byte("value"),
 				Offset: int64(i),
 			}
-			lb.AddLogEntryToBuffer(entry)
+			if err := lb.AddLogEntryToBuffer(entry); err != nil {
+				t.Errorf("Failed to add log entry: %v", err)
+				return
+			}
 			time.Sleep(1 * time.Millisecond) // Simulate production rate
 		}
 		producerDone <- true
@@ -130,7 +133,10 @@ func TestBackwardSeeksWhileProducing(t *testing.T) {
 				Data:   []byte("value"),
 				Offset: int64(i),
 			}
-			lb.AddLogEntryToBuffer(entry)
+			if err := lb.AddLogEntryToBuffer(entry); err != nil {
+				t.Errorf("Failed to add log entry: %v", err)
+				return
+			}
 			time.Sleep(1 * time.Millisecond)
 		}
 		producerDone <- true
@@ -216,7 +222,9 @@ func TestHighConcurrencyReads(t *testing.T) {
 			Data:   []byte("value"),
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry: %v", err)
+		}
 	}
 
 	// Start many concurrent readers at different offsets
@@ -286,7 +294,9 @@ func TestRepeatedReadsAtSameOffset(t *testing.T) {
 			Data:   []byte("value"),
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry: %v", err)
+		}
 	}
 
 	// Read the same offset multiple times concurrently
diff --git a/weed/util/log_buffer/log_read_stateless_test.go b/weed/util/log_buffer/log_read_stateless_test.go
index 948a929ba..6c9206eb4 100644
--- a/weed/util/log_buffer/log_read_stateless_test.go
+++ b/weed/util/log_buffer/log_read_stateless_test.go
@@ -45,7 +45,9 @@ func TestReadMessagesAtOffset_SingleMessage(t *testing.T) {
 		Data:   []byte("value1"),
 		Offset: 0,
 	}
-	lb.AddLogEntryToBuffer(entry)
+	if err := lb.AddLogEntryToBuffer(entry); err != nil {
+		t.Fatalf("Failed to add log entry to buffer: %v", err)
+	}
 
 	// Read from offset 0
 	messages, nextOffset, _, endOfPartition, err := lb.ReadMessagesAtOffset(0, 10, 1024)
@@ -82,7 +84,9 @@ func TestReadMessagesAtOffset_MultipleMessages(t *testing.T) {
 			Data:   []byte("value"),
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry to buffer: %v", err)
+		}
 	}
 
 	// Read from offset 0, max 3 messages
@@ -118,7 +122,9 @@ func TestReadMessagesAtOffset_StartFromMiddle(t *testing.T) {
 			Data:   []byte("value"),
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry to buffer: %v", err)
+		}
 	}
 
 	// Read from offset 5
@@ -155,7 +161,9 @@ func TestReadMessagesAtOffset_MaxBytesLimit(t *testing.T) {
 			Data:   make([]byte, 100), // 100 bytes
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry to buffer: %v", err)
+		}
 	}
 
 	// Request with max 250 bytes (should get ~2 messages)
@@ -186,7 +194,9 @@ func TestReadMessagesAtOffset_ConcurrentReads(t *testing.T) {
 			Data:   []byte("value"),
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry to buffer: %v", err)
+		}
 	}
 
 	// Start 10 concurrent readers at different offsets
@@ -238,7 +248,9 @@ func TestReadMessagesAtOffset_FutureOffset(t *testing.T) {
 			Data:   []byte("value"),
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry to buffer: %v", err)
+		}
 	}
 
 	// Try to read from offset 10 (future)
@@ -269,7 +281,9 @@ func TestWaitForDataWithTimeout_DataAvailable(t *testing.T) {
 		Data:   []byte("value"),
 		Offset: 0,
 	}
-	lb.AddLogEntryToBuffer(entry)
+	if err := lb.AddLogEntryToBuffer(entry); err != nil {
+		t.Fatalf("Failed to add log entry to buffer: %v", err)
+	}
 
 	// Wait for data at offset 0 (should return immediately)
 	dataAvailable := lb.WaitForDataWithTimeout(0, 100)
@@ -321,7 +335,9 @@ func TestWaitForDataWithTimeout_DataArrives(t *testing.T) {
 		Data:   []byte("value"),
 		Offset: 0,
 	}
-	lb.AddLogEntryToBuffer(entry)
+	if err := lb.AddLogEntryToBuffer(entry); err != nil {
+		t.Fatalf("Failed to add log entry to buffer: %v", err)
+	}
 
 	// Wait for result
 	<-done
@@ -349,7 +365,9 @@ func TestGetHighWaterMark(t *testing.T) {
 			Data:   []byte("value"),
 			Offset: int64(i),
 		}
-		lb.AddLogEntryToBuffer(entry)
+		if err := lb.AddLogEntryToBuffer(entry); err != nil {
+			t.Fatalf("Failed to add log entry to buffer: %v", err)
+		}
 	}
 
 	// HWM should be 5 (next offset to write, not last written offset)
diff --git a/weed/util/log_buffer/log_read_test.go b/weed/util/log_buffer/log_read_test.go
index f01e2912a..802dcdacf 100644
--- a/weed/util/log_buffer/log_read_test.go
+++ b/weed/util/log_buffer/log_read_test.go
@@ -171,7 +171,9 @@ func TestLoopProcessLogDataWithOffset_WithData(t *testing.T) {
 	}
 
 	for _, msg := range testMessages {
-		logBuffer.AddToBuffer(msg)
+		if err := logBuffer.AddToBuffer(msg); err != nil {
+			t.Fatalf("Failed to add message to buffer: %v", err)
+		}
 	}
 
 	receivedCount := 0
diff --git a/weed/util/log_buffer/sealed_buffer.go b/weed/util/log_buffer/sealed_buffer.go
index 397dab1d4..109cb3862 100644
--- a/weed/util/log_buffer/sealed_buffer.go
+++ b/weed/util/log_buffer/sealed_buffer.go
@@ -51,16 +51,20 @@ func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte,
 	return oldMemBuffer.buf
 }
 
-func (mb *MemBuffer) locateByTs(lastReadTime time.Time) (pos int) {
+func (mb *MemBuffer) locateByTs(lastReadTime time.Time) (pos int, err error) {
 	lastReadTs := lastReadTime.UnixNano()
 	for pos < len(mb.buf) {
-		size, t := readTs(mb.buf, pos)
+		size, t, readErr := readTs(mb.buf, pos)
+		if readErr != nil {
+			// Return error if buffer is corrupted
+			return 0, fmt.Errorf("locateByTs: buffer corruption at pos %d: %w", pos, readErr)
+		}
 		if t > lastReadTs {
-			return
+			return pos, nil
 		}
 		pos += size + 4
 	}
-	return len(mb.buf)
+	return len(mb.buf), nil
 }
 
 func (mb *MemBuffer) String() string {