Repository: liftbridge-io/liftbridge
Pull request #59 (2019-04-29 01:19:53)

commitlog: ensure index has enough room for all entries on expansion
I ran into an index out of range panic in `Index.ReadAt` while experimenting with liftbridge. (Really cool project, thanks for creating it!)

Found that the index is not expanded in `Index.writeAt` when `position < size && position + len(p) > size`. In this case not all the entries are written to the index but `position` is updated as though they had been, resulting in `position > size`.

Also added a couple test for the initial size before realizing the `Index.InitializePosition` handles updating the position to the correct point on creation. I left them in since they could be useful, but can remove them if you'd like.

Thanks again!


Patch:
diff --git a/server/commitlog/index.go b/server/commitlog/index.go
index e2ce44e..000e173 100644
--- a/server/commitlog/index.go
+++ b/server/commitlog/index.go
@@ -178,9 +178,12 @@ func (idx *Index) ReadAt(p []byte, offset int64) (n int, err error) {
 
 func (idx *Index) writeAt(p []byte, offset int64) (n int) {
 	// Check if we need to expand the index file.
-	if offset >= idx.size {
+	if pSize := int64(len(p)); offset+pSize >= idx.size {
 		// Expand the index file.
 		newSize := roundDown(idx.size+idx.bytes, entryWidth)
+		if newSize < offset+pSize {
+			newSize = idx.size + pSize
+		}
 		err := idx.file.Truncate(newSize)
 		if err != nil {
 			panic(errors.Wrap(err, "failed to expand index file"))


Test patch:
diff --git a/server/commitlog/index_test.go b/server/commitlog/index_test.go
new file mode 100644
index 0000000..dfe361d
--- /dev/null
+++ b/server/commitlog/index_test.go
@@ -0,0 +1,80 @@
+package commitlog
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIndexInitialSize(t *testing.T) {
+	dir := tempDir(t)
+	defer os.RemoveAll(dir)
+
+	// Create a new index.
+	idx, err := NewIndex(options{path: dir + "test.idx"})
+	require.NoError(t, err)
+	entry, err := idx.InitializePosition()
+	require.NoError(t, err)
+	require.Nil(t, entry)
+
+	// Verify the index is preallocated and the position is 0.
+	require.Equal(t, int64(10*1024*1024), idx.size)
+	require.Equal(t, int64(0), idx.position)
+
+	// Verify the recorded size matches the file size.
+	finfo, err := idx.file.Stat()
+	require.NoError(t, err)
+	require.Equal(t, idx.size, finfo.Size())
+}
+
+func TestIndexExistingSize(t *testing.T) {
+	dir := tempDir(t)
+	defer os.RemoveAll(dir)
+
+	// Create a new index.
+	idx, err := NewIndex(options{path: dir + "test.idx"})
+	require.NoError(t, err)
+	entry, err := idx.InitializePosition()
+	require.NoError(t, err)
+	require.Nil(t, entry)
+
+	// Write some entries.
+	err = idx.writeEntries([]*Entry{
+		{}, {}, {}, {},
+	})
+	require.NoError(t, err)
+
+	// Close the index so that it will be truncated.
+	err = idx.Close()
+	require.NoError(t, err)
+
+	// Reopen the index and verify the size and position are correct.
+	idx, err = NewIndex(options{path: dir + "test.idx"})
+	require.NoError(t, err)
+	require.Equal(t, idx.size, int64(entryWidth*4))
+	require.Equal(t, idx.size, idx.position)
+}
+
+func TestIndexExpansion(t *testing.T) {
+	dir := tempDir(t)
+	defer os.RemoveAll(dir)
+
+	// Create an index with enough room for a single entry.
+	idx, err := NewIndex(options{path: dir + "test.idx", bytes: entryWidth})
+	require.NoError(t, err)
+	entry, err := idx.InitializePosition()
+	require.NoError(t, err)
+	require.Nil(t, entry)
+
+	// Write two entries.
+	writeEntry := Entry{Offset: 123, Timestamp: 456, Position: 789, Size: 987}
+	err = idx.writeEntries([]*Entry{{}, &writeEntry})
+	require.NoError(t, err)
+
+	// Check the second entry can be retrieved.
+	var readEntry Entry
+	err = idx.ReadEntryAtLogOffset(&readEntry, 1)
+	require.NoError(t, err)
+	require.Equal(t, writeEntry, readEntry)
+}