fix: fix hang on bad datafile

diskcache/drop.go

@@ -7,6 +7,11 @@ package diskcache
 
 import "os"
 
+const (
+	reasonExceedCapacity = "exceed-max-capacity"
+	reasonBadDataFile    = "bad-data-file"
+)
+
 func (c *DiskCache) dropBatch() error {
 	c.rwlock.Lock()
 	defer c.rwlock.Unlock()
@@ -34,7 +39,7 @@ func (c *DiskCache) dropBatch() error {
 
 		c.dataFiles = c.dataFiles[1:]
 
-		droppedBatchVec.WithLabelValues(c.path).Inc()
+		droppedBatchVec.WithLabelValues(c.path, reasonExceedCapacity).Inc()
 		droppedBytesVec.WithLabelValues(c.path).Add(float64(fi.Size()))
 		datafilesVec.WithLabelValues(c.path).Set(float64(len(c.dataFiles)))
 		sizeVec.WithLabelValues(c.path).Set(float64(c.size))

diskcache/drop_test.go

@@ -49,11 +49,17 @@ func TestDropBatch(t *T.T) {
 	mfs, err := reg.Gather()
 	assert.NoError(t, err)
 
-	t.Logf("\n%s", metrics.MetricFamily2Text(mfs))
+	m := metrics.GetMetricOnLabels(mfs,
+		"diskcache_dropped_total",
+		c.path,
+		reasonExceedCapacity)
+
+	require.NotNil(t, m, "got metrics\n%s", metrics.MetricFamily2Text(mfs))
 
-	m := metrics.GetMetricOnLabels(mfs, "diskcache_dropped_total", c.path)
-	require.NotNil(t, m)
-	assert.Equal(t, float64(1), m.GetCounter().GetValue())
+	assert.Equal(t,
+		float64(1),
+		m.GetCounter().GetValue(),
+		"got metrics\n%s", metrics.MetricFamily2Text(mfs))
 
 	t.Cleanup(func() {
 		assert.NoError(t, c.Close())

diskcache/get.go

@@ -15,6 +15,32 @@ import (
 // Fn is the handler to eat cache from diskcache.
 type Fn func([]byte) error
 
+func (c *DiskCache) switchNextFile() error {
+	if c.curReadfile != "" {
+		if err := c.removeCurrentReadingFile(); err != nil {
+			return fmt.Errorf("removeCurrentReadingFile: %w", err)
+		}
+	}
+
+	// clear .pos
+	if !c.noPos {
+		if err := c.pos.reset(); err != nil {
+			return err
+		}
+	}
+
+	// reopen next file to read
+	return c.doSwitchNextFile()
+}
+
+func (c *DiskCache) skipBadFile() error {
+	defer func() {
+		droppedBatchVec.WithLabelValues(c.path, reasonBadDataFile).Inc()
+	}()
+
+	return c.switchNextFile()
+}
+
 // Get fetch new data from disk cache, then passing to fn
 // if any error occurred during call fn, the reading data is
 // dropped, and will not read again.
@@ -67,32 +93,26 @@ retry:
 	}
 
 	hdr := make([]byte, dataHeaderLen)
-	if n, err = c.rfd.Read(hdr); err != nil {
-		return fmt.Errorf("rfd.Read(%s): %w", c.curReadfile, err)
-	} else if n != dataHeaderLen {
-		return ErrBadHeader
+	if n, err = c.rfd.Read(hdr); err != nil || n != dataHeaderLen {
+		//
+		// On bad datafile, just ignore and delete the file.
+		//
+		if err = c.skipBadFile(); err != nil {
+			return err
+		}
+
+		goto retry // read next new file to save another Get() calling.
 	}
 
+	// how many bytes of current data?
 	nbytes = int(binary.LittleEndian.Uint32(hdr[0:]))
 
 	if uint32(nbytes) == EOFHint { // EOF
-		if err = c.removeCurrentReadingFile(); err != nil {
-			return fmt.Errorf("removeCurrentReadingFile: %w", err)
-		}
-
-		// clear .pos
-		if !c.noPos {
-			if err = c.pos.reset(); err != nil {
-				return err
-			}
-		}
-
-		// reopen next file to read
-		if err = c.switchNextFile(); err != nil {
-			return err
+		if err := c.switchNextFile(); err != nil {
+			return fmt.Errorf("switchNextFile: %w", err)
 		}
 
-		goto retry // read next new file
+		goto retry // read next new file to save another Get() calling.
 	}
 
 	databuf := make([]byte, nbytes)

diskcache/get_test.go

@@ -17,7 +17,85 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestDropInvalidDataFile(t *T.T) {
+	t.Run(`get-on-0bytes-data-file`, func(t *T.T) {
+		p := t.TempDir()
+		c, err := Open(WithPath(p))
+		require.NoError(t, err)
+
+		// put some data and rotate 10 datafiles
+		data := make([]byte, 100)
+		for i := 0; i < 10; i++ {
+			assert.NoError(t, c.Put(data))
+			assert.NoError(t, c.rotate())
+
+			// destroy the datafile
+			if i%2 == 0 {
+				assert.NoError(t, os.Truncate(c.dataFiles[i], 0))
+			}
+		}
+
+		assert.Len(t, c.datafiles, 10)
+
+		for {
+			err := c.Get(func(get []byte) error {
+				// switch to 2nd file
+				assert.Equal(t, data, get)
+				return nil
+			})
+
+			if err != nil {
+				require.ErrorIs(t, err, ErrEOF)
+				break
+			}
+		}
+
+		reg := prometheus.NewRegistry()
+		register(reg)
+		mfs, err := reg.Gather()
+		require.NoError(t, err)
+
+		assert.Equalf(t, float64(5),
+			metrics.GetMetricOnLabels(mfs,
+				"diskcache_dropped_total",
+				c.path,
+				reasonBadDataFile,
+			).GetCounter().GetValue(),
+			"got metrics\n%s", metrics.MetricFamily2Text(mfs))
+	})
+}
+
 func TestFallbackOnError(t *T.T) {
+
+	t.Run(`get-erro-on-EOF`, func(t *T.T) {
+
+		p := t.TempDir()
+		c, err := Open(WithPath(p))
+		require.NoError(t, err)
+
+		// put some data
+		data := make([]byte, 100)
+		assert.NoError(t, c.Put(data))
+
+		assert.NoError(t, c.rotate())
+
+		require.NoError(t, c.Get(func(_ []byte) error {
+			return nil // ignore the data
+		}))
+
+		err = c.Get(func(_ []byte) error {
+			assert.True(t, 1 == 2) // should not been here
+			return nil
+		})
+
+		assert.ErrorIs(t, err, ErrEOF)
+		t.Logf("get: %s", err)
+
+		if errors.Is(err, ErrEOF) {
+			t.Logf("we should ignore the error")
+		}
+	})
+
 	t.Run(`fallback-on-error`, func(t *T.T) {
 		ResetMetrics()
 
@@ -49,7 +127,7 @@ func TestFallbackOnError(t *T.T) {
 		require.NoError(t, err)
 
 		assert.Equalf(t, float64(1),
-			metrics.GetMetricOnLabels(mfs, "diskcache_seek_back", c.path).GetCounter().GetValue(),
+			metrics.GetMetricOnLabels(mfs, "diskcache_seek_back_total", c.path).GetCounter().GetValue(),
 			"got metrics\n%s", metrics.MetricFamily2Text(mfs))
 
 		t.Cleanup(func() {

diskcache/metric.go

@@ -70,7 +70,7 @@ func setupMetrics() {
 			Name:      "dropped_total",
 			Help:      "Dropped files during Put() when capacity reached.",
 		},
-		[]string{"path"},
+		[]string{"path", "reason"},
 	)
 
 	rotateVec = prometheus.NewCounterVec(

diskcache/put_test.go

@@ -251,9 +251,12 @@ func TestPutOnCapacityReached(t *T.T) {
 		mfs, err := reg.Gather()
 		require.NoError(t, err)
 
-		t.Logf("\n%s", metrics.MetricFamily2Text(mfs))
-		m := metrics.GetMetricOnLabels(mfs, "diskcache_dropped_total", c.path)
-		require.NotNil(t, m)
+		m := metrics.GetMetricOnLabels(mfs,
+			"diskcache_dropped_total",
+			c.path,
+			reasonExceedCapacity)
+
+		require.NotNil(t, m, "got metrics:\n%s", metrics.MetricFamily2Text(mfs))
 		assert.True(t, m.GetCounter().GetValue() > 0.0)
 
 		t.Cleanup(func() {
@@ -310,10 +313,12 @@ func TestPutOnCapacityReached(t *T.T) {
 
 		mfs, err := reg.Gather()
 		require.NoError(t, err)
-		t.Logf("\n%s", metrics.MetricFamily2Text(mfs))
 
-		m := metrics.GetMetricOnLabels(mfs, "diskcache_dropped_total", c.path)
-		require.NotNil(t, m)
+		m := metrics.GetMetricOnLabels(mfs,
+			"diskcache_dropped_total",
+			c.path,
+			reasonExceedCapacity)
+		require.NotNil(t, m, "got metrics:\n%s", metrics.MetricFamily2Text(mfs))
 
 		assert.True(t, m.GetCounter().GetValue() > 0.0)
 

diskcache/rotate.go

@@ -103,12 +103,14 @@ func (c *DiskCache) removeCurrentReadingFile() error {
 		c.rfd = nil
 	}
 
-	if fi, err := os.Stat(c.curReadfile); err == nil {
-		c.size -= (fi.Size() - dataHeaderLen) // EOF bytes do not counted in size
-	}
+	if fi, err := os.Stat(c.curReadfile); err == nil { // file exist
+		if fi.Size() > dataHeaderLen {
+			c.size -= (fi.Size() - dataHeaderLen) // EOF bytes do not counted in size
+		}
 
-	if err := os.Remove(c.curReadfile); err != nil {
-		return fmt.Errorf("removeCurrentReadingFile: %s: %w", c.curReadfile, err)
+		if err := os.Remove(c.curReadfile); err != nil {
+			return fmt.Errorf("removeCurrentReadingFile: %q: %w", c.curReadfile, err)
+		}
 	}
 
 	if len(c.dataFiles) > 0 {

diskcache/switch.go

@@ -55,7 +55,7 @@ func (c *DiskCache) loadUnfinishedFile() error {
 }
 
 // open next read file.
-func (c *DiskCache) switchNextFile() error {
+func (c *DiskCache) doSwitchNextFile() error {
 	c.rwlock.Lock()
 	defer c.rwlock.Unlock()