Repository: liftbridge-io/liftbridge
Pull request #244 (2020-08-13 21:17:10)

Fix message retention bug
Fix bug where index out of range panic happens due to a logic error
around message retention rules.


Patch:
diff --git a/server/commitlog/delete_cleaner.go b/server/commitlog/delete_cleaner.go
index a235882..fb13d3a 100644
--- a/server/commitlog/delete_cleaner.go
+++ b/server/commitlog/delete_cleaner.go
@@ -80,6 +80,7 @@ func (c *deleteCleaner) noRetentionLimits() bool {
 }
 
 func (c *deleteCleaner) applyMessagesLimit(segments []*segment) ([]*segment, error) {
+	// We must retain at least the active segment.
 	if len(segments) <= 1 {
 		return segments, nil
 	}
@@ -87,12 +88,13 @@ func (c *deleteCleaner) applyMessagesLimit(segments []*segment) ([]*segment, err
 	// We start at the most recent segment and work our way backwards until we
 	// meet the retention size.
 	var (
-		cleanedSegments = make([]*segment, 0, len(segments))
-		totalMessages   int64
+		lastSeg         = segments[len(segments)-1]
+		cleanedSegments = []*segment{lastSeg}
+		totalMessages   = lastSeg.MessageCount()
 	)
 
 	var i int
-	for i = len(segments) - 1; i > -1; i-- {
+	for i = len(segments) - 2; i > -1; i-- {
 		s := segments[i]
 		totalMessages += s.MessageCount()
 		if totalMessages > c.Retention.Messages {
@@ -116,6 +118,11 @@ func (c *deleteCleaner) applyMessagesLimit(segments []*segment) ([]*segment, err
 }
 
 func (c *deleteCleaner) applyBytesLimit(segments []*segment) ([]*segment, error) {
+	// We must retain at least the active segment.
+	if len(segments) <= 1 {
+		return segments, nil
+	}
+
 	// We start at the most recent segment and work our way backwards until we
 	// meet the retention size.
 	var (
@@ -124,26 +131,23 @@ func (c *deleteCleaner) applyBytesLimit(segments []*segment) ([]*segment, error)
 		totalBytes      = lastSeg.Position()
 	)
 
-	if len(segments) > 1 {
-		var i int
-		for i = len(segments) - 2; i > -1; i-- {
-			s := segments[i]
-			totalBytes += s.Position()
-			if totalBytes > c.Retention.Bytes {
-				break
-			}
-			cleanedSegments = append([]*segment{s}, cleanedSegments...)
+	var i int
+	for i = len(segments) - 2; i > -1; i-- {
+		s := segments[i]
+		totalBytes += s.Position()
+		if totalBytes > c.Retention.Bytes {
+			break
 		}
-		if i > -1 {
-			for ; i > -1; i-- {
-				// TODO: There is an edge case here where we fail partway
-				// through deletion. We will delete some segments but return an
-				// error. This should probably mark segments for deletion,
-				// remove them from the read path, and then delete them
-				// asynchronously.
-				if err := segments[i].Delete(); err != nil {
-					return nil, err
-				}
+		cleanedSegments = append([]*segment{s}, cleanedSegments...)
+	}
+	if i > -1 {
+		for ; i > -1; i-- {
+			// TODO: There is an edge case here where we fail partway through
+			// deletion. We will delete some segments but return an error. This
+			// should probably mark segments for deletion, remove them from the
+			// read path, and then delete them asynchronously.
+			if err := segments[i].Delete(); err != nil {
+				return nil, err
 			}
 		}
 	}
@@ -153,7 +157,7 @@ func (c *deleteCleaner) applyBytesLimit(segments []*segment) ([]*segment, error)
 
 func (c *deleteCleaner) applyAgeLimit(segments []*segment) ([]*segment, error) {
 	// We must retain at least the active segment.
-	if len(segments) == 1 {
+	if len(segments) <= 1 {
 		return segments, nil
 	}
 
diff --git a/website/yarn.lock b/website/yarn.lock
index 1c39c2b..850e54b 100644
--- a/website/yarn.lock
+++ b/website/yarn.lock
@@ -5133,9 +5133,9 @@ prepend-http@^2.0.0:
   integrity sha1-6SQ0v6XqjBn0HN/UAddBo8gZ2Jc=
 
 prismjs@^1.17.1:
-  version "1.21.0"
-  resolved "https://registry.yarnpkg.com/prismjs/-/prismjs-1.21.0.tgz#36c086ec36b45319ec4218ee164c110f9fc015a3"
-  integrity sha512-uGdSIu1nk3kej2iZsLyDoJ7e9bnPzIgY0naW/HdknGj61zScaprVEVGHrPoXqI+M9sP0NDnTK2jpkvmldpuqDw==
+  version "1.19.0"
+  resolved "https://registry.yarnpkg.com/prismjs/-/prismjs-1.19.0.tgz#713afbd45c3baca4b321569f2df39e17e729d4dc"
+  integrity sha512-IVFtbW9mCWm9eOIaEkNyo2Vl4NnEifis2GQ7/MLRG5TQe6t+4Sj9J5QWI9i3v+SS43uZBlCAOn+zYTVYQcPXJw==
   optionalDependencies:
     clipboard "^2.0.0"
 


Test patch:
diff --git a/server/commitlog/delete_cleaner_test.go b/server/commitlog/delete_cleaner_test.go
index aaaa717..9dc1a33 100644
--- a/server/commitlog/delete_cleaner_test.go
+++ b/server/commitlog/delete_cleaner_test.go
@@ -118,6 +118,33 @@ func TestDeleteCleanerMessages(t *testing.T) {
 	}
 }
 
+// Ensure Clean deletes segments to maintain the messages limit but keeps at
+// least the active segment.
+func TestDeleteCleanerMessagesKeepActiveSegment(t *testing.T) {
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts.Retention.Messages = 5
+	cleaner := newDeleteCleaner(opts)
+	dir := tempDir(t)
+	defer remove(t, dir)
+
+	segs := []*segment{
+		createSegment(t, dir, 0, 128),
+		createSegment(t, dir, 10, 128),
+	}
+	offset := int64(0)
+	for _, seg := range segs {
+		for i := 0; i < 10; i++ {
+			writeToSegment(t, seg, offset, []byte("blah"))
+			offset++
+		}
+	}
+
+	actual, err := cleaner.Clean(segs)
+	require.NoError(t, err)
+	require.Len(t, actual, 1)
+	require.Equal(t, int64(10), actual[0].BaseOffset)
+}
+
 // Ensure Clean is a no-op when there are segments and a messages limit but the
 // segments don't exceed the limit.
 func TestDeleteCleanerMessagesBelowLimit(t *testing.T) {