lib/persistentqueue: skip zero-length blocks as corrupt data in readBlock

Zero-length blocks indicate file corruption. Valid blocks should always
have non-zero length. Previously PR
https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6241 worked
around this at the client level by silently skipping such blocks. Move
the check into readBlock, consistent with how other corrupted headers
are handled there, and remove the now-redundant empty-block guards from
the client.

Follow-up on discussion
https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10932#discussion_r3264924234

app/vmagent/remotewrite/client.go

@@ -311,11 +311,6 @@ func (c *client) runWorker() {
 		if !ok {
 			return
 		}
-		if len(block) == 0 {
-			// skip empty data blocks from sending
-			// see https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6241
-			continue
-		}
 		go func() {
 			startTime := time.Now()
 			ch <- c.sendBlock(block)
@@ -530,10 +525,6 @@ func (c *client) drainInMemoryQueue(stopCtx context.Context, block []byte) {
 			// In this case it is guaranteed that fq will be empty
 			return
 		}
-		if len(block) == 0 {
-			// skip empty data blocks from sending
-			continue
-		}
 
 		// at this stage c.stopCh should be closed
 		// so sendBlock function should not perform retries

lib/persistentqueue/persistentqueue.go

@@ -485,6 +485,14 @@ again:
 		}
 		goto again
 	}
+	// see https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6241
+	if blockLen == 0 {
+		logger.Errorf("skipping corrupted %q, since zero block size is read from it", q.readerPath)
+		if err := q.skipBrokenChunkFile(); err != nil {
+			return dst, err
+		}
+		goto again
+	}
 	if blockLen > q.maxBlockSize {
 		logger.Errorf("skipping corrupted %q, since too big block size is read from it: %d bytes; cannot exceed %d bytes", q.readerPath, blockLen, q.maxBlockSize)
 		if err := q.skipBrokenChunkFile(); err != nil {